From 4ba75937c3ce4a1a803259ad93b6f1ba56e4345a Mon Sep 17 00:00:00 2001 From: Gerard Louis Recinto Date: Wed, 21 Oct 2026 19:59:31 -0700 Subject: [PATCH 1/3] Add optional JetStream publishing and a durable pull consumer to the NATS adapter --- CHANGELOG.md | 2 + adapters/nats/bridge.go | 167 ++++++++++++++++++- adapters/nats/bridge_test.go | 305 +++++++++++++++++++++++++++++++++++ adapters/nats/doc.go | 17 ++ adapters/nats/subscribe.go | 100 ++++++++++++ 5 files changed, 587 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 625c3de7c..af59cae0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,8 +2,10 @@ ## Unreleased +- `adapters/nats` can store barrier decisions in JetStream. `VerifyBridge.WithJetStream` creates the stream or checks that an existing one captures the bridge's subject, and publishes each decision asynchronously, so the barrier's answer is never delayed or changed by a publish failure; failures go to `OnPublishError`, and `Flush` waits for pending acknowledgements. `CheckAndCommitIdempotent` publishes with a message ID built from the workflow and idempotency key, so a retry is stored once inside the stream's duplicate window. `SubscribeJetStream` reads them back through a durable pull consumer with explicit acknowledgement, `AckWait` and `MaxDeliver`, so a consumer that was offline catches up. Without `WithJetStream` the bridge behaves exactly as before. - Go is bumped to 1.26.10 and 1.27.3, which clears GO-2026-6809 in `crypto/x509` that the govulncheck and Trivy image scans now report against 1.26.9 and 1.27.2. + ## v5.11.1 - The release build for the Linux and Windows libraries pinned Go 1.26.8 and failed against the Go 1.26.9 requirement, so the v5.11.0 packages were not built. This release fixes the pin. It carries everything listed under v5.11.0. diff --git a/adapters/nats/bridge.go b/adapters/nats/bridge.go index 0688d2e5f..a98a1edf7 100644 --- a/adapters/nats/bridge.go +++ b/adapters/nats/bridge.go @@ -1,12 +1,15 @@ package nats import ( + "context" "encoding/json" "errors" "fmt" + "strings" "time" "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" "github.com/sharedcode/joltrin/v5/verify" ) @@ -80,6 +83,10 @@ type VerifyBridge struct { // simply drops a publish failure, exactly like fire-and-forget metrics // or logging, rather than letting it surface as a barrier error. onPublishError func(error) + + // js is set only by WithJetStream. When nil, publish uses core NATS + // exactly as before. + js jetstream.JetStream } // NewVerifyBridge returns a VerifyBridge that runs barrier checks against @@ -113,7 +120,7 @@ func (b *VerifyBridge) Subject() string { // BarrierDecision, and returns wf.CheckSafety's exact result. func (b *VerifyBridge) CheckSafety(trace *verify.Trace, next verify.StepID) error { err := b.wf.CheckSafety(trace, next) - b.publish(next, false, err) + b.publish(next, false, "", err) return err } @@ -121,7 +128,7 @@ func (b *VerifyBridge) CheckSafety(trace *verify.Trace, next verify.StepID) erro // resulting BarrierDecision, and returns wf.CheckAndCommit's exact result. func (b *VerifyBridge) CheckAndCommit(trace *verify.Trace, next verify.StepID) error { err := b.wf.CheckAndCommit(trace, next) - b.publish(next, false, err) + b.publish(next, false, "", err) return err } @@ -131,11 +138,11 @@ func (b *VerifyBridge) CheckAndCommit(trace *verify.Trace, next verify.StepID) e // returns wf.CheckAndCommitIdempotent's exact result. func (b *VerifyBridge) CheckAndCommitIdempotent(trace *verify.Trace, next verify.StepID, idempotencyKey string) (replayed bool, err error) { replayed, err = b.wf.CheckAndCommitIdempotent(trace, next, idempotencyKey) - b.publish(next, replayed, err) + b.publish(next, replayed, idempotencyKey, err) return replayed, err } -func (b *VerifyBridge) publish(step verify.StepID, replayed bool, err error) { +func (b *VerifyBridge) publish(step verify.StepID, replayed bool, idempotencyKey string, err error) { ev := BarrierDecision{ Time: time.Now().UTC(), Workflow: b.workflow, @@ -164,7 +171,159 @@ func (b *VerifyBridge) publish(step verify.StepID, replayed bool, err error) { return } + if b.js != nil { + b.publishJetStream(data, idempotencyKey) + return + } + if err := b.nc.Publish(b.subject, data); err != nil && b.onPublishError != nil { b.onPublishError(fmt.Errorf("nats: publish barrier decision: %w", err)) } } + +// DefaultStream is the JetStream stream WithJetStream uses when +// JetStreamOptions.Stream is empty. It captures every subject a +// VerifyBridge publishes to, whatever the workflow name. +const DefaultStream = "JOLTRIN_VERIFY" + +// defaultStreamSubjects is the subject filter DefaultStream is created with. +const defaultStreamSubjects = "joltrin.verify.>" + +// JetStreamOptions configures the stream WithJetStream creates or checks. +// The zero value is valid. +type JetStreamOptions struct { + // Stream is the stream name. Empty means DefaultStream. + Stream string + // Replicas is the replica count used only when the stream is created. + // Zero means 1. + Replicas int + // MaxAge, if set, is how long the stream keeps a decision. Used only + // when the stream is created. Zero keeps decisions until removed. + MaxAge time.Duration + // Duplicates is the window in which a repeated Nats-Msg-Id is stored + // once. Used only when the stream is created. Zero means the server + // default (two minutes). + Duplicates time.Duration +} + +// WithJetStream makes the bridge publish each BarrierDecision to a +// JetStream stream instead of plain NATS, so decisions are stored and a +// consumer that was offline can read them later (see SubscribeJetStream). +// It returns the bridge for chaining. +// +// The stream is created if it does not exist. If it exists, it is checked +// to capture the bridge's subject and is otherwise left untouched; a +// mismatch is an error rather than a silent change to someone else's +// stream. +// +// Publishing stays off the barrier's path. The decision is made first and +// returned unchanged, and the publish is asynchronous: a failure, including +// a missing stream or a lost acknowledgement, is reported through +// OnPublishError and never to the caller of CheckSafety, CheckAndCommit, or +// CheckAndCommitIdempotent. Call Flush before shutdown to wait for pending +// acknowledgements. +// +// CheckAndCommitIdempotent publishes with a Nats-Msg-Id built from the +// workflow, step, and idempotency key, so a retry is stored once inside the +// stream's duplicate window. CheckSafety and CheckAndCommit have no +// caller-supplied identity, so their decisions are published without an ID +// and are not de-duplicated. +func (b *VerifyBridge) WithJetStream(ctx context.Context, opts JetStreamOptions) (*VerifyBridge, error) { + js, err := jetstream.New(b.nc, jetstream.WithPublishAsyncErrHandler(func(_ jetstream.JetStream, _ *nats.Msg, err error) { + if b.onPublishError != nil { + b.onPublishError(fmt.Errorf("nats: jetstream publish barrier decision: %w", err)) + } + })) + if err != nil { + return nil, fmt.Errorf("nats: open jetstream: %w", err) + } + + name := opts.Stream + if name == "" { + name = DefaultStream + } + + stream, err := js.Stream(ctx, name) + switch { + case errors.Is(err, jetstream.ErrStreamNotFound): + replicas := opts.Replicas + if replicas == 0 { + replicas = 1 + } + subjects := []string{defaultStreamSubjects} + if !subjectCovered(subjects, b.subject) { + subjects = []string{b.subject} + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{ + Name: name, + Subjects: subjects, + Replicas: replicas, + MaxAge: opts.MaxAge, + Duplicates: opts.Duplicates, + }); err != nil { + return nil, fmt.Errorf("nats: create stream %q: %w", name, err) + } + case err != nil: + return nil, fmt.Errorf("nats: look up stream %q: %w", name, err) + default: + if !subjectCovered(stream.CachedInfo().Config.Subjects, b.subject) { + return nil, fmt.Errorf("nats: stream %q does not capture subject %q", name, b.subject) + } + } + + b.js = js + return b, nil +} + +// Flush waits until every asynchronous JetStream publish has been +// acknowledged or has failed, or ctx is done. It does nothing for a bridge +// without WithJetStream. +func (b *VerifyBridge) Flush(ctx context.Context) error { + if b.js == nil { + return nil + } + select { + case <-b.js.PublishAsyncComplete(): + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (b *VerifyBridge) publishJetStream(data []byte, idempotencyKey string) { + var opts []jetstream.PublishOpt + if idempotencyKey != "" { + opts = append(opts, jetstream.WithMsgID(b.workflow+"/"+idempotencyKey)) + } + if _, err := b.js.PublishMsgAsync(&nats.Msg{Subject: b.subject, Data: data}, opts...); err != nil && b.onPublishError != nil { + b.onPublishError(fmt.Errorf("nats: jetstream publish barrier decision: %w", err)) + } +} + +// subjectCovered reports whether subject matches any of the stream subject +// patterns, using NATS wildcards: "*" for one token and a trailing ">" for +// one or more. +func subjectCovered(patterns []string, subject string) bool { + subjTokens := strings.Split(subject, ".") + for _, p := range patterns { + pt := strings.Split(p, ".") + match := true + for i, tok := range pt { + if tok == ">" { + match = i < len(subjTokens) + break + } + if i >= len(subjTokens) || (tok != "*" && tok != subjTokens[i]) { + match = false + break + } + if i == len(pt)-1 && len(subjTokens) != len(pt) { + match = false + } + } + if match { + return true + } + } + return false +} diff --git a/adapters/nats/bridge_test.go b/adapters/nats/bridge_test.go index 06a2cff93..38b0c7479 100644 --- a/adapters/nats/bridge_test.go +++ b/adapters/nats/bridge_test.go @@ -1,13 +1,16 @@ package nats import ( + "context" "encoding/json" + "errors" "sync" "testing" "time" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" "github.com/sharedcode/joltrin/v5/verify" ) @@ -348,3 +351,305 @@ func TestVerifyBridge_PublishFailureDoesNotChangeDecision(t *testing.T) { t.Fatalf("OnPublishError called %d times, want 2", publishErrs) } } + +// startJetStreamServer is startTestServer with JetStream enabled on a temp +// store directory. +func startJetStreamServer(t *testing.T) *nats.Conn { + t.Helper() + + srv, err := server.NewServer(&server.Options{ + Host: "127.0.0.1", + Port: -1, + NoLog: true, + NoSigs: true, + JetStream: true, + StoreDir: t.TempDir(), + }) + if err != nil { + t.Fatalf("start embedded NATS server: %v", err) + } + go srv.Start() + if !srv.ReadyForConnections(5 * time.Second) { + t.Fatal("embedded NATS server did not become ready") + } + t.Cleanup(srv.Shutdown) + + nc, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatalf("connect to embedded NATS server: %v", err) + } + t.Cleanup(nc.Close) + return nc +} + +func jsBridge(t *testing.T, nc *nats.Conn, opts JetStreamOptions) *VerifyBridge { + t.Helper() + bridge, err := NewVerifyBridge(dropProdWorkflow(t), nc, "prod-db-rollout").WithJetStream(context.Background(), opts) + if err != nil { + t.Fatalf("WithJetStream: %v", err) + } + return bridge +} + +func streamMsgs(t *testing.T, nc *nats.Conn, name string) uint64 { + t.Helper() + js, err := jetstream.New(nc) + if err != nil { + t.Fatalf("jetstream.New: %v", err) + } + stream, err := js.Stream(context.Background(), name) + if err != nil { + t.Fatalf("Stream: %v", err) + } + info, err := stream.Info(context.Background()) + if err != nil { + t.Fatalf("Stream.Info: %v", err) + } + return info.State.Msgs +} + +// TestJetStream_StoredWhileConsumerOffline publishes before any consumer +// exists, then starts a durable consumer and expects every decision. +func TestJetStream_StoredWhileConsumerOffline(t *testing.T) { + nc := startJetStreamServer(t) + bridge := jsBridge(t, nc, JetStreamOptions{}) + trace := verify.NewTrace() + + if err := bridge.CheckAndCommit(trace, "drop_prod_db"); err == nil { + t.Fatal("drop_prod_db should be blocked before a backup is validated") + } + if err := bridge.CheckAndCommit(trace, "take_backup"); err != nil { + t.Fatalf("take_backup: %v", err) + } + if err := bridge.Flush(context.Background()); err != nil { + t.Fatalf("Flush: %v", err) + } + if got := streamMsgs(t, nc, DefaultStream); got != 2 { + t.Fatalf("stream holds %d messages, want 2", got) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var ( + mu sync.Mutex + got []BarrierDecision + ) + errc := make(chan error, 1) + go func() { + errc <- SubscribeJetStream(ctx, nc, bridge.Subject(), ConsumerOptions{Durable: "late"}, func(ev BarrierDecision) error { + mu.Lock() + got = append(got, ev) + mu.Unlock() + return nil + }) + }() + + deadline := time.Now().Add(5 * time.Second) + for { + mu.Lock() + n := len(got) + mu.Unlock() + if n == 2 { + break + } + if time.Now().After(deadline) { + t.Fatalf("got %d decisions, want 2", n) + } + time.Sleep(20 * time.Millisecond) + } + mu.Lock() + if got[0].Allowed || got[0].Step != "drop_prod_db" || !got[1].Allowed || got[1].Step != "take_backup" { + t.Errorf("unexpected decisions: %+v", got) + } + mu.Unlock() + + cancel() + if err := <-errc; !errors.Is(err, context.Canceled) { + t.Fatalf("SubscribeJetStream returned %v, want context.Canceled", err) + } +} + +// TestJetStream_DuplicateIdempotentPublishStoredOnce retries one idempotent +// call; the retry is published with the same message ID and stored once. +func TestJetStream_DuplicateIdempotentPublishStoredOnce(t *testing.T) { + nc := startJetStreamServer(t) + bridge := jsBridge(t, nc, JetStreamOptions{}) + trace := verify.NewTrace() + + if replayed, err := bridge.CheckAndCommitIdempotent(trace, "take_backup", "k1"); err != nil || replayed { + t.Fatalf("first call: replayed=%v err=%v", replayed, err) + } + if replayed, err := bridge.CheckAndCommitIdempotent(trace, "take_backup", "k1"); err != nil || !replayed { + t.Fatalf("retry: replayed=%v err=%v", replayed, err) + } + if err := bridge.Flush(context.Background()); err != nil { + t.Fatalf("Flush: %v", err) + } + if got := streamMsgs(t, nc, DefaultStream); got != 1 { + t.Fatalf("stream holds %d messages after a retry, want 1", got) + } +} + +// TestJetStream_UnackedMessageRedelivered returns an error from the handler +// on the first delivery and expects the server to redeliver after AckWait. +func TestJetStream_UnackedMessageRedelivered(t *testing.T) { + nc := startJetStreamServer(t) + bridge := jsBridge(t, nc, JetStreamOptions{}) + if err := bridge.CheckAndCommit(verify.NewTrace(), "take_backup"); err != nil { + t.Fatalf("take_backup: %v", err) + } + if err := bridge.Flush(context.Background()); err != nil { + t.Fatalf("Flush: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var ( + mu sync.Mutex + attempts int + ) + delivered := make(chan struct{}) + go func() { + _ = SubscribeJetStream(ctx, nc, bridge.Subject(), ConsumerOptions{Durable: "retry", AckWait: 200 * time.Millisecond, MaxDeliver: 3}, func(BarrierDecision) error { + mu.Lock() + defer mu.Unlock() + attempts++ + if attempts == 1 { + return errors.New("not ready") + } + select { + case <-delivered: + default: + close(delivered) + } + return nil + }) + }() + + select { + case <-delivered: + case <-time.After(5 * time.Second): + t.Fatal("message was not redelivered after AckWait") + } + mu.Lock() + defer mu.Unlock() + if attempts != 2 { + t.Errorf("attempts = %d, want 2", attempts) + } +} + +// TestJetStream_PublishFailureDoesNotChangeDecision removes the stream so +// the publish fails, then checks the barrier's answer is the one verify +// gives on its own and the failure reached OnPublishError. +func TestJetStream_PublishFailureDoesNotChangeDecision(t *testing.T) { + nc := startJetStreamServer(t) + failed := make(chan error, 4) + bridge := jsBridge(t, nc, JetStreamOptions{}).OnPublishError(func(err error) { + select { + case failed <- err: + default: + } + }) + + js, err := jetstream.New(nc) + if err != nil { + t.Fatalf("jetstream.New: %v", err) + } + if err := js.DeleteStream(context.Background(), DefaultStream); err != nil { + t.Fatalf("DeleteStream: %v", err) + } + + wantErr := dropProdWorkflow(t).CheckAndCommit(verify.NewTrace(), "drop_prod_db") + gotErr := bridge.CheckAndCommit(verify.NewTrace(), "drop_prod_db") + var want, got *verify.Violation + if !errors.As(wantErr, &want) || !errors.As(gotErr, &got) || got.Rule != want.Rule || gotErr.Error() != wantErr.Error() { + t.Fatalf("bridge decision %v differs from verify's %v", gotErr, wantErr) + } + + select { + case err := <-failed: + if err == nil { + t.Fatal("OnPublishError called with nil") + } + case <-time.After(5 * time.Second): + t.Fatal("publish failure was not reported to OnPublishError") + } +} + +func TestJetStream_ExistingStreamMustCaptureSubject(t *testing.T) { + nc := startJetStreamServer(t) + js, err := jetstream.New(nc) + if err != nil { + t.Fatalf("jetstream.New: %v", err) + } + if _, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "OTHER", Subjects: []string{"other.>"}}); err != nil { + t.Fatalf("CreateStream: %v", err) + } + _, err = NewVerifyBridge(dropProdWorkflow(t), nc, "prod-db-rollout").WithJetStream(context.Background(), JetStreamOptions{Stream: "OTHER"}) + if err == nil { + t.Fatal("WithJetStream accepted a stream that does not capture the bridge's subject") + } +} + +func TestSubscribeJetStream_RequiresDurableAndTermsMalformed(t *testing.T) { + nc := startJetStreamServer(t) + if err := SubscribeJetStream(context.Background(), nc, DefaultSubject, ConsumerOptions{}, nil); err == nil { + t.Fatal("SubscribeJetStream accepted an empty Durable") + } + + bridge := jsBridge(t, nc, JetStreamOptions{}) + // A plain request to a stream subject is stored by the stream with no + // payload check, so this puts an unparseable message ahead of the real + // decision. + if _, err := nc.Request(bridge.Subject(), []byte("not json"), 2*time.Second); err != nil { + t.Fatalf("publish malformed message: %v", err) + } + if err := bridge.CheckAndCommit(verify.NewTrace(), "take_backup"); err != nil { + t.Fatalf("take_backup: %v", err) + } + if err := bridge.Flush(context.Background()); err != nil { + t.Fatalf("Flush: %v", err) + } + if n := streamMsgs(t, nc, DefaultStream); n != 2 { + t.Fatalf("stream holds %d messages, want 2 (one malformed, one valid)", n) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + got := make(chan BarrierDecision, 2) + go func() { + _ = SubscribeJetStream(ctx, nc, bridge.Subject(), ConsumerOptions{Durable: "mixed"}, func(ev BarrierDecision) error { + got <- ev + return nil + }) + }() + select { + case ev := <-got: + if ev.Step != "take_backup" { + t.Errorf("handler saw %+v, want the take_backup decision", ev) + } + case <-time.After(5 * time.Second): + t.Fatal("valid decision behind a malformed message was not delivered") + } +} + +func TestSubjectCovered(t *testing.T) { + cases := []struct { + patterns []string + subject string + want bool + }{ + {[]string{"joltrin.verify.>"}, DefaultSubject, true}, + {[]string{"joltrin.verify.>"}, "joltrin.verify.a.decision", true}, + {[]string{"joltrin.verify.*.decision"}, "joltrin.verify.a.decision", true}, + {[]string{"joltrin.verify.*.decision"}, DefaultSubject, false}, + {[]string{"joltrin.verify.decision"}, "joltrin.verify.decision.extra", false}, + {[]string{"other.>"}, DefaultSubject, false}, + {[]string{"joltrin.>"}, "joltrin", false}, + } + for _, c := range cases { + if got := subjectCovered(c.patterns, c.subject); got != c.want { + t.Errorf("subjectCovered(%v, %q) = %v, want %v", c.patterns, c.subject, got, c.want) + } + } +} diff --git a/adapters/nats/doc.go b/adapters/nats/doc.go index 4ffb9c0b0..f0f486e0a 100644 --- a/adapters/nats/doc.go +++ b/adapters/nats/doc.go @@ -30,4 +30,21 @@ // and every barrier decision made through bridge is also published as a // BarrierDecision event to bridge's subject. See examples/verify_barrier_nats // for a full runnable demonstration, including a subscriber. +// +// # JetStream +// +// Plain NATS drops an event nobody is listening for. For decisions that must +// be kept and replayed, opt in with WithJetStream, which stores each decision +// in a stream, and read them back with SubscribeJetStream, a durable pull +// consumer with explicit acknowledgement: +// +// bridge, err := nats.NewVerifyBridge(wf, nc, "prod-db-rollout"). +// WithJetStream(ctx, nats.JetStreamOptions{}) +// // later, possibly after a restart: +// err = nats.SubscribeJetStream(ctx, nc, bridge.Subject(), +// nats.ConsumerOptions{Durable: "audit"}, func(ev nats.BarrierDecision) error { ... }) +// +// The decision is made and returned before any publish, and a JetStream +// failure is reported only through OnPublishError. A caller that never calls +// WithJetStream gets the plain NATS behavior described above, unchanged. package nats diff --git a/adapters/nats/subscribe.go b/adapters/nats/subscribe.go index 8dd5f27cd..071d8d201 100644 --- a/adapters/nats/subscribe.go +++ b/adapters/nats/subscribe.go @@ -1,10 +1,14 @@ package nats import ( + "context" "encoding/json" + "errors" "fmt" + "time" "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" ) // Subscribe subscribes to subject on nc and calls handler with each @@ -32,3 +36,99 @@ func Subscribe(nc *nats.Conn, subject string, handler func(BarrierDecision)) (*n } return sub, nil } + +// ConsumerOptions configures SubscribeJetStream. Durable is required; the +// other fields have defaults. +type ConsumerOptions struct { + // Stream is the stream to read. Empty means DefaultStream. + Stream string + // Durable is the consumer name. The server remembers its position, so + // a consumer that restarts with the same name resumes where it left + // off. + Durable string + // AckWait is how long the server waits for an acknowledgement before + // redelivering. Zero means 30 seconds. + AckWait time.Duration + // MaxDeliver caps delivery attempts per message. Zero means 5. + MaxDeliver int + // Batch is how many messages to request at a time. Zero means 10. + Batch int +} + +// SubscribeJetStream reads BarrierDecision events for subject from a +// JetStream stream through a durable pull consumer with explicit +// acknowledgement, and calls handler for each. It blocks until ctx is done +// and returns ctx.Err() then, or an earlier setup or fetch error. +// +// A message is acknowledged when handler returns nil. If handler returns an +// error the message is left unacknowledged, so the server redelivers it +// after AckWait, up to MaxDeliver attempts. A message that does not +// unmarshal as a BarrierDecision is terminated rather than redelivered, +// matching Subscribe's rule that one malformed message must not stop the +// rest. +// +// Because the consumer is durable, decisions published while no consumer +// was running are delivered when one starts. +func SubscribeJetStream(ctx context.Context, nc *nats.Conn, subject string, opts ConsumerOptions, handler func(BarrierDecision) error) error { + if opts.Durable == "" { + return errors.New("nats: SubscribeJetStream requires ConsumerOptions.Durable") + } + stream := opts.Stream + if stream == "" { + stream = DefaultStream + } + ackWait := opts.AckWait + if ackWait == 0 { + ackWait = 30 * time.Second + } + maxDeliver := opts.MaxDeliver + if maxDeliver == 0 { + maxDeliver = 5 + } + batch := opts.Batch + if batch == 0 { + batch = 10 + } + + js, err := jetstream.New(nc) + if err != nil { + return fmt.Errorf("nats: open jetstream: %w", err) + } + cons, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{ + Durable: opts.Durable, + FilterSubject: subject, + AckPolicy: jetstream.AckExplicitPolicy, + AckWait: ackWait, + MaxDeliver: maxDeliver, + }) + if err != nil { + return fmt.Errorf("nats: consumer %q on stream %q: %w", opts.Durable, stream, err) + } + + for { + if err := ctx.Err(); err != nil { + return err + } + msgs, err := cons.Fetch(batch, jetstream.FetchMaxWait(time.Second)) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return fmt.Errorf("nats: fetch from consumer %q: %w", opts.Durable, err) + } + for msg := range msgs.Messages() { + var ev BarrierDecision + if err := json.Unmarshal(msg.Data(), &ev); err != nil { + _ = msg.Term() + continue + } + if err := handler(ev); err != nil { + continue + } + _ = msg.Ack() + } + if err := msgs.Error(); err != nil && ctx.Err() == nil && !errors.Is(err, nats.ErrTimeout) { + return fmt.Errorf("nats: fetch from consumer %q: %w", opts.Durable, err) + } + } +} From 8d13d8a354f14657f0e302de05e9d6c77159bc96 Mon Sep 17 00:00:00 2001 From: Gerard Louis Recinto Date: Wed, 21 Oct 2026 21:21:32 -0700 Subject: [PATCH 2/3] Include the step in the JetStream message ID and make the ID unambiguous --- CHANGELOG.md | 4 +--- adapters/nats/bridge.go | 22 ++++++++++++++++++--- adapters/nats/bridge_test.go | 38 ++++++++++++++++++++++++++++++++++++ 3 files changed, 58 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index af59cae0f..a1f3713a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,9 +2,7 @@ ## Unreleased -- `adapters/nats` can store barrier decisions in JetStream. `VerifyBridge.WithJetStream` creates the stream or checks that an existing one captures the bridge's subject, and publishes each decision asynchronously, so the barrier's answer is never delayed or changed by a publish failure; failures go to `OnPublishError`, and `Flush` waits for pending acknowledgements. `CheckAndCommitIdempotent` publishes with a message ID built from the workflow and idempotency key, so a retry is stored once inside the stream's duplicate window. `SubscribeJetStream` reads them back through a durable pull consumer with explicit acknowledgement, `AckWait` and `MaxDeliver`, so a consumer that was offline catches up. Without `WithJetStream` the bridge behaves exactly as before. -- Go is bumped to 1.26.10 and 1.27.3, which clears GO-2026-6809 in `crypto/x509` that the govulncheck and Trivy image scans now report against 1.26.9 and 1.27.2. - +- `adapters/nats` can store barrier decisions in JetStream. `VerifyBridge.WithJetStream` creates the stream or checks that an existing one captures the bridge's subject, and publishes each decision asynchronously, so the barrier's answer is never delayed or changed by a publish failure; failures go to `OnPublishError`, and `Flush` waits for pending acknowledgements. `CheckAndCommitIdempotent` publishes with a message ID built from the workflow, step, and idempotency key, so a retry is stored once inside the stream's duplicate window. `SubscribeJetStream` reads them back through a durable pull consumer with explicit acknowledgement, `AckWait` and `MaxDeliver`, so a consumer that was offline catches up. Without `WithJetStream` the bridge behaves exactly as before. ## v5.11.1 diff --git a/adapters/nats/bridge.go b/adapters/nats/bridge.go index a98a1edf7..a7be7e473 100644 --- a/adapters/nats/bridge.go +++ b/adapters/nats/bridge.go @@ -2,6 +2,8 @@ package nats import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -172,7 +174,7 @@ func (b *VerifyBridge) publish(step verify.StepID, replayed bool, idempotencyKey } if b.js != nil { - b.publishJetStream(data, idempotencyKey) + b.publishJetStream(data, step, idempotencyKey) return } @@ -290,16 +292,30 @@ func (b *VerifyBridge) Flush(ctx context.Context) error { } } -func (b *VerifyBridge) publishJetStream(data []byte, idempotencyKey string) { +func (b *VerifyBridge) publishJetStream(data []byte, step verify.StepID, idempotencyKey string) { var opts []jetstream.PublishOpt if idempotencyKey != "" { - opts = append(opts, jetstream.WithMsgID(b.workflow+"/"+idempotencyKey)) + opts = append(opts, jetstream.WithMsgID(messageID(b.workflow, step, idempotencyKey))) } if _, err := b.js.PublishMsgAsync(&nats.Msg{Subject: b.subject, Data: data}, opts...); err != nil && b.onPublishError != nil { b.onPublishError(fmt.Errorf("nats: jetstream publish barrier decision: %w", err)) } } +// messageID returns the Nats-Msg-Id for an idempotent decision: a SHA-256 of +// the workflow, step, and idempotency key, each length-prefixed so no choice +// of values can make two different triples produce the same input. The step +// is part of it because verify refuses to reuse a key for a different step; +// without the step, that refusal would share an ID with the original +// decision and be dropped as a duplicate. +func messageID(workflow string, step verify.StepID, key string) string { + h := sha256.New() + for _, part := range []string{workflow, string(step), key} { + fmt.Fprintf(h, "%d:%s", len(part), part) + } + return hex.EncodeToString(h.Sum(nil)) +} + // subjectCovered reports whether subject matches any of the stream subject // patterns, using NATS wildcards: "*" for one token and a trailing ">" for // one or more. diff --git a/adapters/nats/bridge_test.go b/adapters/nats/bridge_test.go index 38b0c7479..9aefc3d3c 100644 --- a/adapters/nats/bridge_test.go +++ b/adapters/nats/bridge_test.go @@ -653,3 +653,41 @@ func TestSubjectCovered(t *testing.T) { } } } + +// TestMessageID_UnambiguousAndStepScoped pins the two properties the +// de-duplication relies on: different inputs never share an ID, including +// inputs that would collide under a plain "/" join, and the step is part of +// the ID. +func TestMessageID_UnambiguousAndStepScoped(t *testing.T) { + if messageID("a/b", "s", "c") == messageID("a", "s", "b/c") { + t.Error("workflow/key boundary is ambiguous") + } + if messageID("w", "s1", "k") == messageID("w", "s2", "k") { + t.Error("step is not part of the message ID") + } + if messageID("w", "s", "k") != messageID("w", "s", "k") { + t.Error("message ID is not stable") + } +} + +// TestJetStream_KeyReusedForOtherStepIsNotDeduplicated reuses an idempotency +// key for a different step. verify refuses it; that refusal must be stored +// as its own event, not dropped as a duplicate of the first decision. +func TestJetStream_KeyReusedForOtherStepIsNotDeduplicated(t *testing.T) { + nc := startJetStreamServer(t) + bridge := jsBridge(t, nc, JetStreamOptions{}) + trace := verify.NewTrace() + + if _, err := bridge.CheckAndCommitIdempotent(trace, "take_backup", "k1"); err != nil { + t.Fatalf("first call: %v", err) + } + if _, err := bridge.CheckAndCommitIdempotent(trace, "validate_backup", "k1"); err == nil { + t.Fatal("reusing a key for another step should be refused") + } + if err := bridge.Flush(context.Background()); err != nil { + t.Fatalf("Flush: %v", err) + } + if got := streamMsgs(t, nc, DefaultStream); got != 2 { + t.Fatalf("stream holds %d messages, want 2 (decision and refusal)", got) + } +} From 480765e56f198482bd0ebe2b36eeb28c22013f49 Mon Sep 17 00:00:00 2001 From: Gerard Louis Recinto Date: Wed, 21 Oct 2026 21:55:54 -0700 Subject: [PATCH 3/3] Keep the default JetStream stream broad and scope custom streams to their own subject --- CHANGELOG.md | 1 + adapters/nats/bridge.go | 5 ++++- adapters/nats/bridge_test.go | 28 ++++++++++++++++++++++++++++ 3 files changed, 33 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a1f3713a8..d901cf40f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Go is bumped to 1.26.10 and 1.27.3, which clears GO-2026-6809 in `crypto/x509` that the govulncheck and Trivy image scans now report against 1.26.9 and 1.27.2. - `adapters/nats` can store barrier decisions in JetStream. `VerifyBridge.WithJetStream` creates the stream or checks that an existing one captures the bridge's subject, and publishes each decision asynchronously, so the barrier's answer is never delayed or changed by a publish failure; failures go to `OnPublishError`, and `Flush` waits for pending acknowledgements. `CheckAndCommitIdempotent` publishes with a message ID built from the workflow, step, and idempotency key, so a retry is stored once inside the stream's duplicate window. `SubscribeJetStream` reads them back through a durable pull consumer with explicit acknowledgement, `AckWait` and `MaxDeliver`, so a consumer that was offline catches up. Without `WithJetStream` the bridge behaves exactly as before. ## v5.11.1 diff --git a/adapters/nats/bridge.go b/adapters/nats/bridge.go index a7be7e473..744b13011 100644 --- a/adapters/nats/bridge.go +++ b/adapters/nats/bridge.go @@ -252,8 +252,11 @@ func (b *VerifyBridge) WithJetStream(ctx context.Context, opts JetStreamOptions) if replicas == 0 { replicas = 1 } + // The default stream is shared by every bridge, so it captures the + // whole joltrin.verify tree. A stream with a custom name holds only + // this bridge's subject, so it cannot overlap one it does not own. subjects := []string{defaultStreamSubjects} - if !subjectCovered(subjects, b.subject) { + if name != DefaultStream { subjects = []string{b.subject} } if _, err := js.CreateStream(ctx, jetstream.StreamConfig{ diff --git a/adapters/nats/bridge_test.go b/adapters/nats/bridge_test.go index 9aefc3d3c..21f7e1ea5 100644 --- a/adapters/nats/bridge_test.go +++ b/adapters/nats/bridge_test.go @@ -691,3 +691,31 @@ func TestJetStream_KeyReusedForOtherStepIsNotDeduplicated(t *testing.T) { t.Fatalf("stream holds %d messages, want 2 (decision and refusal)", got) } } + +// TestJetStream_DefaultStreamSharedCustomStreamScoped checks the stream +// subjects WithJetStream creates: the default stream takes every workflow, +// a custom stream only its own bridge's subject. +func TestJetStream_DefaultStreamSharedCustomStreamScoped(t *testing.T) { + nc := startJetStreamServer(t) + ctx := context.Background() + for _, wfName := range []string{"one", "two"} { + bridge, err := NewVerifyBridge(dropProdWorkflow(t), nc, wfName).WithJetStream(ctx, JetStreamOptions{}) + if err != nil { + t.Fatalf("WithJetStream(%q): %v", wfName, err) + } + if err := bridge.CheckAndCommit(verify.NewTrace(), "take_backup"); err != nil { + t.Fatalf("take_backup: %v", err) + } + if err := bridge.Flush(ctx); err != nil { + t.Fatalf("Flush: %v", err) + } + } + if got := streamMsgs(t, nc, DefaultStream); got != 2 { + t.Fatalf("default stream holds %d messages from two workflows, want 2", got) + } + + custom := NewVerifyBridge(dropProdWorkflow(t), nc, "three") + if _, err := custom.WithJetStream(ctx, JetStreamOptions{Stream: "ONLY_THREE"}); err == nil { + t.Fatal("custom stream for a subject the default stream already captures should be rejected by the server as overlapping") + } +}