diff --git a/CHANGELOG.md b/CHANGELOG.md index 625c3de7c..d901cf40f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,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 0688d2e5f..744b13011 100644 --- a/adapters/nats/bridge.go +++ b/adapters/nats/bridge.go @@ -1,12 +1,17 @@ package nats import ( + "context" + "crypto/sha256" + "encoding/hex" "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 +85,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 +122,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 +130,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 +140,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 +173,176 @@ func (b *VerifyBridge) publish(step verify.StepID, replayed bool, err error) { return } + if b.js != nil { + b.publishJetStream(data, step, 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 + } + // 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 name != DefaultStream { + 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, step verify.StepID, idempotencyKey string) { + var opts []jetstream.PublishOpt + if 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. +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..21f7e1ea5 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,371 @@ 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) + } + } +} + +// 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) + } +} + +// 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") + } +} 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) + } + } +}