From 97028c33b5de888df67dc716ce1ba328ef4d49a3 Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 10:17:09 +0000 Subject: [PATCH] feat(events): skip in-process writers when a consumer is live Keep OnOutbox and approval hooks registered while dispatcher mode is off, but no-op them as soon as the matching durable consumer owns the visible write so a later switch cannot dual-write. Co-authored-by: Cursor --- .../ops}/2026-09-15-event-e1c-projection.md | 0 .doc/system-design/module-boundaries.yaml | 1 + internal/app/app.go | 18 ++++++++++++++++++ internal/app/docnotify/subscriber.go | 11 ++++++++++- internal/events/dispatcher.go | 11 +++++++++++ internal/server/workflow_approval_hook.go | 4 ++++ 6 files changed, 44 insertions(+), 1 deletion(-) rename .doc/{ops/short-term => archive/ops}/2026-09-15-event-e1c-projection.md (100%) diff --git a/.doc/ops/short-term/2026-09-15-event-e1c-projection.md b/.doc/archive/ops/2026-09-15-event-e1c-projection.md similarity index 100% rename from .doc/ops/short-term/2026-09-15-event-e1c-projection.md rename to .doc/archive/ops/2026-09-15-event-e1c-projection.md diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index d4fb184c..cc3ee7c2 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -161,6 +161,7 @@ imports: - docgov - docquery - docrevision + - events - freedoc - intent - notification diff --git a/internal/app/app.go b/internal/app/app.go index d4979476..aefc2024 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -153,6 +153,9 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { Stageflow: stages, } rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: svc.DocGov()} + rt.DocNotify.SkipWrite = func(ctx context.Context) bool { + return rt.WriterLive(ctx, events.ConsumerNotify) + } envName := "" if cfg != nil { envName = cfg.Env @@ -193,7 +196,22 @@ func (r *Runtime) bindWorkflowApprovalHook() { r.subscribers++ } +// WriterLive reports whether the durable consumer already owns the visible write. +func (r *Runtime) WriterLive(ctx context.Context, consumer string) bool { + if r == nil || r.Biz == nil { + return false + } + env := "" + if r.Cfg != nil { + env = r.Cfg.Env + } + return events.WriterLive(ctx, r.Biz.Pool(), env, consumer) +} + func (r *Runtime) onWorkflowDocumentApproved(ctx context.Context, doc *docgov.DocumentRecord) { + if r.WriterLive(ctx, events.ConsumerWorkflowAdvance) { + return + } if doc == nil || doc.TaskKey == "" || doc.ProjectID <= 0 || r.Inst == nil { return } diff --git a/internal/app/docnotify/subscriber.go b/internal/app/docnotify/subscriber.go index 2025e864..1e516c85 100644 --- a/internal/app/docnotify/subscriber.go +++ b/internal/app/docnotify/subscriber.go @@ -19,6 +19,9 @@ type DocumentLookup interface { type Subscriber struct { Notify notificationEmitter Docs DocumentLookup + // SkipWrite is true when the durable notify consumer is live. + // Bind stays registered while mode=off; the closure must not dual-write after switch. + SkipWrite func(ctx context.Context) bool } type notificationEmitter interface { @@ -37,7 +40,13 @@ func Bind(rev *docrevision.Service, sub *Subscriber) bool { // Handle maps one document outbox event onto a permission-safe inbox notification. func (s *Subscriber) Handle(ctx context.Context, eventType string, payload map[string]any) error { - if s == nil || s.Notify == nil || !s.Notify.Available() { + if s == nil { + return nil + } + if s.SkipWrite != nil && s.SkipWrite(ctx) { + return nil + } + if s.Notify == nil || !s.Notify.Available() { return nil } docID, _ := payload["doc_id"].(string) diff --git a/internal/events/dispatcher.go b/internal/events/dispatcher.go index c4a71dbf..3939239b 100644 --- a/internal/events/dispatcher.go +++ b/internal/events/dispatcher.go @@ -302,6 +302,17 @@ func LoadMode(ctx context.Context, pool *pgxpool.Pool, environment, consumer str return mode, rev, err } +// WriterLive is true only when the durable consumer already owns the visible write. +// Closures must no-op in that window; off/shadow keep the in-process hook as the unique writer. +// A missing pool or mode row is treated as not-live so HTTP notify/advance keep working. +func WriterLive(ctx context.Context, pool *pgxpool.Pool, environment, consumer string) bool { + if pool == nil { + return false + } + mode, _, err := LoadMode(ctx, pool, environment, consumer) + return err == nil && mode == ModeLive +} + func SetMode(ctx context.Context, pool *pgxpool.Pool, environment, consumer, mode string) error { if consumer == ConsumerFAV && mode != ModeOff { return fmt.Errorf("%w: fav", ErrDisabledConsumer) diff --git a/internal/server/workflow_approval_hook.go b/internal/server/workflow_approval_hook.go index a38db097..ef5e5f6f 100644 --- a/internal/server/workflow_approval_hook.go +++ b/internal/server/workflow_approval_hook.go @@ -4,6 +4,7 @@ import ( "context" "chat-agent-system/internal/docgov" + "chat-agent-system/internal/events" "chat-agent-system/internal/notification" ) @@ -15,6 +16,9 @@ func (s *Server) registerWorkflowApprovalHook() { } func (s *Server) onWorkflowDocumentApproved(ctx context.Context, doc *docgov.DocumentRecord) { + if s.rt != nil && s.rt.WriterLive(ctx, events.ConsumerWorkflowAdvance) { + return + } if doc == nil || doc.TaskKey == "" || doc.ProjectID <= 0 || s.instSvc == nil { return }