weichen
29f14ac5f2
Details (original message): Add multi-instance groundwork: blobstore seam for artifacts, PG LISTEN/NOTIFY event fanout, and per-tenant rate limiting with native Redis backend
109 lines
2.9 KiB
Go
109 lines
2.9 KiB
Go
// Package pgbus 是 PG 档的事件跨实例扇出（阶段 3）：
|
||
// 落盘事务提交后 NOTIFY agent_events '<conv>|<lastSeq>'；每实例一个 LISTEN 连接，
|
||
// 收到通知按 seq 从库取回事件并转发本地 bus（SSE 以 seq 去重，双投无害）。
|
||
// 多实例部署契约：run 在其归属实例内存中执行，会话流量需按会话亲和路由
|
||
//（网关/客户端择点）；本包只解决“其他实例的订阅者”可见性。
|
||
package pgbus
|
||
|
||
import (
|
||
"context"
|
||
"log"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
core "agent-system-core"
|
||
"agent-system-backend/internal/bus"
|
||
"agent-system-backend/internal/store"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
)
|
||
|
||
const channel = "agent_events"
|
||
|
||
// Notifier 落盘后通知（任意连接池连接可用）。
|
||
type Notifier struct{ st *store.Store }
|
||
|
||
func NewNotifier(st *store.Store) *Notifier { return &Notifier{st: st} }
|
||
|
||
// Notify 在 AppendEvents 成功后调用：一批事件一条通知（载荷=会话|末 seq）。
|
||
func (n *Notifier) Notify(conversationID string, evs []core.StreamEvent) {
|
||
if len(evs) == 0 {
|
||
return
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||
defer cancel()
|
||
payload := conversationID + "|" + strconv.FormatInt(evs[len(evs)-1].Seq, 10)
|
||
if _, err := n.st.NotifyPG(ctx, channel, payload); err != nil {
|
||
log.Printf("[pgbus] notify %s: %v", conversationID, err)
|
||
}
|
||
}
|
||
|
||
// Listener 每实例一个：LISTEN → 按 seq 取回 → 本地 bus 转发。
|
||
type Listener struct {
|
||
conn *pgx.Conn
|
||
st *store.Store
|
||
bus *bus.Bus
|
||
done chan struct{}
|
||
}
|
||
|
||
// Start 建立监听连接并开始转发循环；返回后即可用，随 ctx 取消停机。
|
||
func Start(ctx context.Context, dsn string, st *store.Store, b *bus.Bus) (*Listener, error) {
|
||
conn, err := pgx.Connect(ctx, dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if _, err := conn.Exec(ctx, "LISTEN "+channel); err != nil {
|
||
conn.Close(ctx)
|
||
return nil, err
|
||
}
|
||
l := &Listener{conn: conn, st: st, bus: b, done: make(chan struct{})}
|
||
go l.loop(ctx)
|
||
return l, nil
|
||
}
|
||
|
||
func (l *Listener) loop(ctx context.Context) {
|
||
defer close(l.done)
|
||
for {
|
||
n, err := l.conn.WaitForNotification(ctx)
|
||
if err != nil {
|
||
if ctx.Err() != nil {
|
||
return // 停机
|
||
}
|
||
log.Printf("[pgbus] wait: %v（1s 后重连）", err)
|
||
time.Sleep(time.Second)
|
||
continue
|
||
}
|
||
conv, seq, ok := parsePayload(n.Payload)
|
||
if !ok {
|
||
continue
|
||
}
|
||
ev, err := l.st.EventAt(conv, seq)
|
||
if err != nil {
|
||
continue // 事件不可见（极早期窗口）——SSE 重放兜底
|
||
}
|
||
l.bus.Publish(ev)
|
||
}
|
||
}
|
||
|
||
func parsePayload(p string) (string, int64, bool) {
|
||
i := strings.LastIndex(p, "|")
|
||
if i < 0 {
|
||
return "", 0, false
|
||
}
|
||
seq, err := strconv.ParseInt(p[i+1:], 10, 64)
|
||
if err != nil {
|
||
return "", 0, false
|
||
}
|
||
return p[:i], seq, true
|
||
}
|
||
|
||
// Close 关闭监听连接。
|
||
func (l *Listener) Close(ctx context.Context) {
|
||
l.conn.Close(ctx)
|
||
select {
|
||
case <-l.done:
|
||
case <-time.After(2 * time.Second):
|
||
}
|
||
}
|