From 4b14649b3aa6034600cfd09f5f943e9dbb48c3d8 Mon Sep 17 00:00:00 2001 From: Stephan D Date: Sat, 12 Sep 2026 14:52:44 +0200 Subject: [PATCH] Complete notification expiry and configurable durable delivery (#36) --- backend/notifications/README.md | 20 +++- backend/notifications/app/config.go | 7 +- backend/notifications/app/run.go | 2 +- backend/notifications/bus/config.go | 38 ++++++++ backend/notifications/bus/config_test.go | 23 +++++ backend/notifications/bus/dispatch.go | 7 +- backend/notifications/config.example.yml | 10 ++ backend/notifications/service/grpc_test.go | 58 +++++++++++ .../notifications/service/integration_test.go | 95 ++++++++++++++++--- backend/notifications/store/mongo.go | 17 +++- backend/notifications/transport/server.go | 6 ++ .../notifications/transport/server_test.go | 9 ++ .../messaging/internal/natscap/adapter.go | 10 +- .../messaging/internal/natscap/admin.go | 29 +++--- .../internal/natscap/subscription_test.go | 12 +++ docs/implementation.md | 17 ++++ 16 files changed, 320 insertions(+), 40 deletions(-) create mode 100644 backend/notifications/bus/config.go create mode 100644 backend/notifications/bus/config_test.go create mode 100644 backend/notifications/service/grpc_test.go diff --git a/backend/notifications/README.md b/backend/notifications/README.md index 5aed36d..843d17d 100644 --- a/backend/notifications/README.md +++ b/backend/notifications/README.md @@ -32,10 +32,16 @@ Use a private authenticated NATS endpoint; `tls://` URLs and credentials in the are supported by the shared driver. Mongo and messaging both use shared `DriverConfig` settings; email settings reject unknown keys. -The service provisions `SXCHNGE_NOTIFICATIONS` (file storage, 48-hour retention, -one replica), subject `sxchnge.notifications.email`, and durable consumer `email`. -This initial setup is for a single-replica deployment. Configure production -replication/provisioning policy in code before deploying to a clustered environment. +The queue uses `SXCHNGE_NOTIFICATIONS`, subject `sxchnge.notifications.email`, +and durable consumer `email`. Configure `queue.stream.replicas` as 1, 3 or 5 to +match the NATS cluster, and `queue.stream.max_age_hours` for retention (48 hours +minimum). File storage and the owned stream/subject names are enforced. +`queue.provisioning.environment` accepts `production` or `non-production`; +`mode: ops` validates existing resources without changing them, `mode: app` +provisions missing resources, and `mode: hybrid` provisions only outside production. +Production retention and consumer-setting drift fail startup instead of being +silently accepted. Provisioning operations honor their startup context. +The example defaults to one replica and app provisioning for local development. Multiple service workers can safely compete for jobs through atomic claims. ## Templates and parameters @@ -54,6 +60,8 @@ remaining status and request fingerprints expire through a seven-day TTL. - Send persists before returning; it does not wait for NATS or the email provider. - Outbox scans every five seconds and republishes safely after a bus failure. + Pending notifications expire and have their rendered content cleared even when + the bus is unavailable or a retry is scheduled later. - Duplicate RPCs and duplicate broker deliveries do not create extra jobs/sends. - Provider HTTP 2xx becomes ACCEPTED. No receipt/webhook or read tracking is claimed. - Explicit rate limiting (429) retries with bounded exponential backoff and @@ -80,7 +88,9 @@ Monitor pending age, UNKNOWN/FAILED states and repeated outbox/worker warnings. `make test` and `make lint-backend` generate interfaces and validate all modules. Notifications tests exercise real disposable MongoDB and NATS containers, concurrent idempotency and claims, lease recovery, retries, transport authentication, rendering -and both provider payloads through local HTTP servers. No live emails are sent. +and both provider payloads through local HTTP servers. An authenticated TLS gRPC +Send/Get test covers the full shared-database → NATS → rendered-email driver path. +No live emails are sent. `make package-notifications-interface` prepares source and Go-client archives. See [provenance](PROVENANCE.md) and [contract usage](interface/README.md). diff --git a/backend/notifications/app/config.go b/backend/notifications/app/config.go index 134db44..8359077 100644 --- a/backend/notifications/app/config.go +++ b/backend/notifications/app/config.go @@ -6,6 +6,7 @@ import ( "os" "strings" + "git.sendico.io/tech/sxchnge/notifications/bus" "git.sendico.io/tech/sxchnge/notifications/email" db "git.sendico.io/tech/sxchnge/shared/database" merrors "git.sendico.io/tech/sxchnge/shared/errors" @@ -14,6 +15,7 @@ import ( ) type Config struct { + Queue bus.Config `yaml:"queue"` Messaging messaging.Config `yaml:"messaging"` Listen string `yaml:"listen"` TLSCert string `yaml:"tls_cert"` @@ -25,7 +27,7 @@ type Config struct { } func Load(path string) (Config, error) { - var c Config + c := Config{Queue: bus.DefaultConfig()} data, err := os.ReadFile(path) if err != nil { return c, merrors.OpaqueWrap(merrors.ErrInvalidArg, err, "cannot read configuration") @@ -41,6 +43,9 @@ func Load(path string) (Config, error) { if c.Listen == "" || c.TLSCert == "" || c.TLSKey == "" || c.Database.Driver != db.Mongo || c.Messaging.Driver != messaging.BBNats || c.Templates == "" || len(c.Callers) == 0 { return c, merrors.InvalidArgument("missing notification configuration") } + if err = c.Queue.Validate(); err != nil { + return c, err + } seen := map[string]bool{} for caller, env := range c.Callers { token := os.Getenv(env) diff --git a/backend/notifications/app/run.go b/backend/notifications/app/run.go index a883c74..da24ff1 100644 --- a/backend/notifications/app/run.go +++ b/backend/notifications/app/run.go @@ -58,7 +58,7 @@ func Run(ctx context.Context, c Config, logger mlogger.Logger) error { } defer rt.Close() svc := service.New(repo, catalog, sender, string(c.Email.Driver), logger) - broker, err := bus.New(startup, rt.Durable, rt.DurableAdmin, repo, svc, logger) + broker, err := bus.New(startup, rt.Durable, rt.DurableAdmin, repo, svc, logger, c.Queue) if err != nil { return err } diff --git a/backend/notifications/bus/config.go b/backend/notifications/bus/config.go new file mode 100644 index 0000000..9cbf43f --- /dev/null +++ b/backend/notifications/bus/config.go @@ -0,0 +1,38 @@ +package bus + +import ( + merrors "git.sendico.io/tech/sxchnge/shared/errors" + "git.sendico.io/tech/sxchnge/shared/messaging/reliable" +) + +// Config composes canonical shared messaging settings for this service's queue. +type Config struct { + Stream reliable.DurableStreamSettings `yaml:"stream"` + Provisioning reliable.DurableProvisioningSettings `yaml:"provisioning"` +} + +func DefaultConfig() Config { + return Config{ + Stream: reliable.DurableStreamSettings{Name: "SXCHNGE_NOTIFICATIONS", Subjects: []string{subject}, Storage: "file", MaxAgeHours: 48, Replicas: 1}, + Provisioning: reliable.DurableProvisioningSettings{Mode: reliable.DurableProvisioningApp, Environment: reliable.DurableEnvironmentNonProduction}, + } +} +func (c Config) Validate() error { + if c.Stream.Name != "SXCHNGE_NOTIFICATIONS" || len(c.Stream.Subjects) != 1 || c.Stream.Subjects[0] != subject || c.Stream.Storage != "file" || c.Stream.MaxAgeHours < 48 || c.Stream.MaxAgeHours > 24*30 { + return merrors.InvalidArgument("invalid notifications stream configuration") + } + if c.Stream.Replicas != 1 && c.Stream.Replicas != 3 && c.Stream.Replicas != 5 { + return merrors.InvalidArgument("stream replicas must be 1, 3 or 5") + } + switch c.Provisioning.Mode { + case reliable.DurableProvisioningApp, reliable.DurableProvisioningOps, reliable.DurableProvisioningHybrid: + default: + return merrors.InvalidArgument("invalid stream provisioning mode") + } + switch c.Provisioning.Environment { + case reliable.DurableEnvironmentProduction, reliable.DurableEnvironmentNonProduction: + default: + return merrors.InvalidArgument("invalid stream provisioning environment") + } + return nil +} diff --git a/backend/notifications/bus/config_test.go b/backend/notifications/bus/config_test.go new file mode 100644 index 0000000..4af477d --- /dev/null +++ b/backend/notifications/bus/config_test.go @@ -0,0 +1,23 @@ +package bus + +import ( + "git.sendico.io/tech/sxchnge/shared/messaging/reliable" + "testing" +) + +func TestQueueConfiguration(t *testing.T) { + config := DefaultConfig() + config.Stream.Replicas = 3 + config.Provisioning.Mode = reliable.DurableProvisioningOps + config.Provisioning.Environment = reliable.DurableEnvironmentProduction + if err := config.Validate(); err != nil { + t.Fatal(err) + } + for _, change := range []func(*Config){func(c *Config) { c.Stream.Replicas = 0 }, func(c *Config) { c.Stream.Storage = "memory" }, func(c *Config) { c.Stream.MaxAgeHours = 1 }, func(c *Config) { c.Provisioning.Mode = "typo" }, func(c *Config) { c.Provisioning.Environment = "typo" }, func(c *Config) { c.Stream.Subjects = []string{"other"} }} { + c := DefaultConfig() + change(&c) + if err := c.Validate(); err == nil { + t.Fatal("invalid queue configuration accepted") + } + } +} diff --git a/backend/notifications/bus/dispatch.go b/backend/notifications/bus/dispatch.go index 93b57b3..37482c0 100644 --- a/backend/notifications/bus/dispatch.go +++ b/backend/notifications/bus/dispatch.go @@ -29,8 +29,11 @@ type Bus struct { logger mlogger.Logger } -func New(ctx context.Context, transport reliable.Transport, admin reliable.Admin, outbox Outbox, handler Handler, logger mlogger.Logger) (*Bus, error) { - settings := reliable.DurableSettings{Enabled: true, Stream: reliable.DurableStreamSettings{Name: "SXCHNGE_NOTIFICATIONS", Subjects: []string{subject}, Storage: "file", MaxAgeHours: 48, Replicas: 1}, Provisioning: reliable.DurableProvisioningSettings{Mode: reliable.DurableProvisioningApp, Environment: reliable.DurableEnvironmentNonProduction}} +func New(ctx context.Context, transport reliable.Transport, admin reliable.Admin, outbox Outbox, handler Handler, logger mlogger.Logger, config Config) (*Bus, error) { + if err := config.Validate(); err != nil { + return nil, err + } + settings := reliable.DurableSettings{Enabled: true, Stream: config.Stream, Provisioning: config.Provisioning} consumer := reliable.DurableConsumerSettings{Durable: "email", Subject: subject, AckWaitSeconds: 60, MaxDeliver: -1, DeliverPolicy: reliable.DurableDeliverAll} if err := admin.EnsureConsumerResources(ctx, settings, consumer); err != nil { return nil, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "configure notification messaging") diff --git a/backend/notifications/config.example.yml b/backend/notifications/config.example.yml index f45a22c..373e531 100644 --- a/backend/notifications/config.example.yml +++ b/backend/notifications/config.example.yml @@ -22,3 +22,13 @@ email: api_key_env: "RESEND_API_KEY" # SENDGRID_API_KEY when driver is sendgrid sender_address: "notifications@example.com" sender_name: "SXCHNGE" + +# Use replicas: 3 and environment: production for a three-node NATS cluster. +# mode: ops validates resources without modifying them; app provisions them. +queue: + stream: + replicas: 1 + max_age_hours: 48 + provisioning: + mode: app + environment: non-production diff --git a/backend/notifications/service/grpc_test.go b/backend/notifications/service/grpc_test.go new file mode 100644 index 0000000..93f7f15 --- /dev/null +++ b/backend/notifications/service/grpc_test.go @@ -0,0 +1,58 @@ +package service_test + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "math/big" + "net" + "testing" + "time" + + pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1" + "git.sendico.io/tech/sxchnge/notifications/service" + "git.sendico.io/tech/sxchnge/notifications/transport" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/metadata" +) + +// Exercise authenticated TLS transport against the real service and storage. +func grpcClient(t *testing.T, svc *service.Service) pb.NotificationsServiceClient { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + template := &x509.Certificate{SerialNumber: big.NewInt(1), NotBefore: time.Now().Add(-time.Minute), NotAfter: time.Now().Add(time.Hour), DNSNames: []string{"localhost"}, KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}} + der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + certificate, err := x509.ParseCertificate(der) + if err != nil { + t.Fatal(err) + } + roots := x509.NewCertPool() + roots.AddCert(certificate) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer(grpc.Creds(credentials.NewTLS(&tls.Config{MinVersion: tls.VersionTLS12, Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}})), grpc.UnaryInterceptor(transport.Authenticate(map[string]string{"product": "product-test-token", "other": "other-test-token"}))) + pb.RegisterNotificationsServiceServer(server, transport.New(svc)) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{MinVersion: tls.VersionTLS12, RootCAs: roots, ServerName: "localhost"}))) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + return pb.NewNotificationsServiceClient(conn) +} +func callerContext(ctx context.Context, caller string) context.Context { + return metadata.NewOutgoingContext(ctx, metadata.Pairs("authorization", "Bearer "+caller+"-test-token")) +} diff --git a/backend/notifications/service/integration_test.go b/backend/notifications/service/integration_test.go index 402de93..1be0f83 100644 --- a/backend/notifications/service/integration_test.go +++ b/backend/notifications/service/integration_test.go @@ -3,6 +3,7 @@ package service_test import ( "context" "errors" + "strings" "sync" "sync/atomic" "testing" @@ -10,6 +11,7 @@ import ( "git.sendico.io/tech/sxchnge/notifications/bus" "git.sendico.io/tech/sxchnge/notifications/email" + pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1" "git.sendico.io/tech/sxchnge/notifications/model" "git.sendico.io/tech/sxchnge/notifications/service" "git.sendico.io/tech/sxchnge/notifications/store" @@ -18,19 +20,30 @@ import ( "git.sendico.io/tech/sxchnge/shared/database/repository" merrors "git.sendico.io/tech/sxchnge/shared/errors" "git.sendico.io/tech/sxchnge/shared/messaging" + "git.sendico.io/tech/sxchnge/shared/messaging/reliable" msgruntime "git.sendico.io/tech/sxchnge/shared/messaging/runtime" "github.com/testcontainers/testcontainers-go" "github.com/testcontainers/testcontainers-go/modules/mongodb" "github.com/testcontainers/testcontainers-go/wait" "go.uber.org/zap" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) type sender struct { - calls atomic.Int32 - state model.State + calls atomic.Int32 + mu sync.Mutex + messages map[string]model.Email + state model.State } -func (s *sender) Send(context.Context, string, model.Email) email.Outcome { +func (s *sender) Send(_ context.Context, id string, message model.Email) email.Outcome { + s.mu.Lock() + if s.messages == nil { + s.messages = map[string]model.Email{} + } + s.messages[id] = message + s.mu.Unlock() s.calls.Add(1) return email.Outcome{State: s.state, Code: "test", RetryAfter: time.Millisecond} } @@ -159,6 +172,25 @@ func TestDurableDelivery(t *testing.T) { if err != nil || retry.State != model.RetryWait || retryDriver.calls.Load() != 1 { t.Fatal(retry, err) } + // Expire both queued and retrying notifications without a connected broker. + for _, state := range []model.State{model.Queued, model.RetryWait} { + request.Key = "offline-expiry-" + string(state) + expired, e := svc.Send(ctx, "product", request) + if e != nil { + t.Fatal(e) + } + patch := repository.Patch().Set(repository.Field("state"), state).Set(repository.Field("expires_at"), time.Now().Add(-time.Minute)).Set(repository.Field("next_attempt_at"), time.Now().Add(time.Hour)) + if e = repo.DBImp.Patch(ctx, expired.ID, patch); e != nil { + t.Fatal(e) + } + if e = repo.Recover(ctx, time.Now()); e != nil { + t.Fatal(e) + } + expired, e = svc.Get(ctx, "product", expired.ID.Hex()) + if e != nil || expired.State != model.Expired || expired.Email.Address != "" { + t.Fatal("offline expiry failed", expired, e) + } + } // Verify the real JetStream publisher/consumer path with an isolated broker. ncContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ContainerRequest: testcontainers.ContainerRequest{Image: "nats:2.11-alpine", ExposedPorts: []string{"4222/tcp"}, Cmd: []string{"-js"}, WaitingFor: wait.ForListeningPort("4222/tcp")}, Started: true}) if err != nil { @@ -175,16 +207,48 @@ func TestDurableDelivery(t *testing.T) { t.Fatal(err) } defer rt.Close() - broker, err := bus.New(ctx, rt.Durable, rt.DurableAdmin, repo, svc, zap.NewNop()) + broker, err := bus.New(ctx, rt.Durable, rt.DurableAdmin, repo, svc, zap.NewNop(), bus.DefaultConfig()) if err != nil { t.Fatal(err) } defer broker.Close() - request.Key = "jetstream" - job, err := svc.Send(ctx, "product", request) + // Production ops mode validates existing resources without silently rewriting drift. + queue := bus.DefaultConfig() + settings := reliable.DurableSettings{Stream: queue.Stream, Provisioning: reliable.DurableProvisioningSettings{Mode: reliable.DurableProvisioningOps, Environment: reliable.DurableEnvironmentProduction}} + if err = rt.DurableAdmin.EnsureProducerResources(ctx, settings); err != nil { + t.Fatal(err) + } + settings.Stream.MaxAgeHours = 72 + if err = rt.DurableAdmin.EnsureProducerResources(ctx, settings); !errors.Is(err, merrors.ErrDataConflict) { + t.Fatal("retention drift accepted", err) + } + settings.Stream.MaxAgeHours = 48 + if err = rt.DurableAdmin.EnsureProducerResources(ctx, settings); err != nil { + t.Fatal("ops mode mutated the stream", err) + } + canceled, stopAdmin := context.WithCancel(ctx) + stopAdmin() + if err = rt.DurableAdmin.EnsureProducerResources(canceled, settings); err == nil { + t.Fatal("canceled provisioning succeeded") + } + + client := grpcClient(t, svc) + rpcCtx := callerContext(ctx, "product") + input := &pb.SendRequest{IdempotencyKey: "jetstream", UserId: "u", TemplateId: "test", TemplateVersion: "1", Parameters: map[string]string{"name": "End to end & escaped"}, Targets: []*pb.Target{{Destination: &pb.Target_Email{Email: &pb.EmailTarget{Address: "user@example.com"}}}}} + if _, e := client.Send(ctx, input); status.Code(e) != codes.Unauthenticated { + t.Fatal("unauthenticated request accepted", e) + } + sent, err := client.Send(rpcCtx, input) if err != nil { t.Fatal(err) } + replay, err := client.Send(rpcCtx, input) + if err != nil || replay.Notification.Id != sent.Notification.Id { + t.Fatal("RPC idempotency failed", err) + } + if _, e := client.Get(callerContext(ctx, "other"), &pb.GetRequest{Id: sent.Notification.Id}); status.Code(e) != codes.NotFound { + t.Fatal("RPC caller isolation failed", e) + } if err = broker.PublishDue(ctx); err != nil { t.Fatal(err) } @@ -194,16 +258,25 @@ func TestDurableDelivery(t *testing.T) { defer func() { stop(); <-done }() deadline := time.Now().Add(10 * time.Second) for { - job, err = svc.Get(ctx, "product", job.ID.Hex()) - if err != nil { - t.Fatal(err) + result, e := client.Get(rpcCtx, &pb.GetRequest{Id: sent.Notification.Id}) + if e != nil { + t.Fatal(e) } - if job.State == model.Accepted { + if strings.Contains(result.String(), "user@example.com") || strings.Contains(result.String(), "escaped") { + t.Fatal("response exposes email content") + } + if result.Notification.Deliveries[0].State == pb.DeliveryState_DELIVERY_STATE_ACCEPTED { break } if time.Now().After(deadline) { - t.Fatal("JetStream job did not complete", job.State) + t.Fatal("JetStream job did not complete", result.Notification.Deliveries[0].State) } time.Sleep(20 * time.Millisecond) } + driver.mu.Lock() + delivered := driver.messages[sent.Notification.Id] + driver.mu.Unlock() + if delivered.HTML != "

End to end & escaped

" || delivered.Subject != "Hello" || delivered.Address != "user@example.com" { + t.Fatal("parameters did not reach driver", delivered) + } } diff --git a/backend/notifications/store/mongo.go b/backend/notifications/store/mongo.go index de86cfd..70bffd3 100644 --- a/backend/notifications/store/mongo.go +++ b/backend/notifications/store/mongo.go @@ -14,6 +14,7 @@ import ( mlogger "git.sendico.io/tech/sxchnge/shared/logging" "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" + "go.uber.org/zap" ) type Store struct { @@ -26,6 +27,7 @@ func New(logger mlogger.Logger, db *mongo.Database) (*Store, error) { for _, index := range []*ri.Definition{ {Name: "unique_caller_key", Keys: []ri.Key{{Field: "caller", Sort: ri.Asc}, {Field: "key", Sort: ri.Asc}}, Unique: true}, {Name: "pending_dispatch", Keys: []ri.Key{{Field: "state", Sort: ri.Asc}, {Field: "next_attempt_at", Sort: ri.Asc}}}, + {Name: "pending_expiry", Keys: []ri.Key{{Field: "state", Sort: ri.Asc}, {Field: "expires_at", Sort: ri.Asc}}}, {Name: "expired_lease", Keys: []ri.Key{{Field: "state", Sort: ri.Asc}, {Field: "lease_until", Sort: ri.Asc}}}, {Name: "purge_notifications", Keys: []ri.Key{{Field: "purge_at", Sort: ri.Asc}}, TTL: &ttl}, } { @@ -69,11 +71,24 @@ func pending(now time.Time) builder.Query { return repository.Query().And(repository.Query().In(repository.Field("state"), model.Queued, model.RetryWait), repository.Query().Comparison(repository.Field("next_attempt_at"), builder.Lte, now)) } +// Expire pending messages independently of bus availability, then fence lost workers. // A lost worker may already have sent its message. Do not automatically replay it. func (s *Store) Recover(ctx context.Context, now time.Time) error { + expired := repository.Query().And(repository.Query().In(repository.Field("state"), model.Queued, model.RetryWait), repository.Query().Comparison(repository.Field("expires_at"), builder.Lte, now)) + expirePatch := repository.Patch().Set(repository.Field("state"), model.Expired).Set(repository.Field("failure_code"), "expired").Set(repository.Field("updatedAt"), now).Set(repository.Field("email"), model.Email{}) + count, err := s.PatchMany(ctx, expired, expirePatch) + if err != nil { + return wrap(err) + } + if count > 0 { + s.Logger.Info("Pending notifications expired", zap.Int("count", count)) + } query := repository.Query().And(repository.Filter("state", model.Sending), repository.Query().Comparison(repository.Field("lease_until"), builder.Lte, now)) patch := repository.Patch().Set(repository.Field("state"), model.Unknown).Set(repository.Field("failure_code"), "worker_outcome_unknown").Set(repository.Field("updatedAt"), now).Set(repository.Field("email"), model.Email{}) - _, err := s.PatchMany(ctx, query, patch) + count, err = s.PatchMany(ctx, query, patch) + if err == nil && count > 0 { + s.Logger.Warn("Notification workers lost their leases", zap.Int("count", count)) + } return wrap(err) } func (s *Store) Claim(ctx context.Context, id string, now time.Time) (*model.Notification, error) { diff --git a/backend/notifications/transport/server.go b/backend/notifications/transport/server.go index 2bb2604..57b1fe7 100644 --- a/backend/notifications/transport/server.go +++ b/backend/notifications/transport/server.go @@ -77,6 +77,12 @@ func Authenticate(tokens map[string]string) grpc.UnaryServerInterceptor { } } func rpcError(err error) error { + if errors.Is(err, context.Canceled) { + return status.Error(codes.Canceled, "notification operation canceled") + } + if errors.Is(err, context.DeadlineExceeded) { + return status.Error(codes.DeadlineExceeded, "notification operation deadline exceeded") + } for _, pair := range []struct { err error code codes.Code diff --git a/backend/notifications/transport/server_test.go b/backend/notifications/transport/server_test.go index 36ee47e..f6cc70b 100644 --- a/backend/notifications/transport/server_test.go +++ b/backend/notifications/transport/server_test.go @@ -64,3 +64,12 @@ func TestGRPCAuthenticationAndMapping(t *testing.T) { t.Fatal(err) } } + +func TestCancellationStatus(t *testing.T) { + if status.Code(rpcError(merrors.OpaqueWrap(merrors.ErrServiceUnavailable, context.Canceled, "storage"))) != codes.Canceled { + t.Fatal("cancellation lost at transport boundary") + } + if status.Code(rpcError(context.DeadlineExceeded)) != codes.DeadlineExceeded { + t.Fatal("deadline lost at transport boundary") + } +} diff --git a/backend/shared/messaging/internal/natscap/adapter.go b/backend/shared/messaging/internal/natscap/adapter.go index 7fdad23..a824513 100644 --- a/backend/shared/messaging/internal/natscap/adapter.go +++ b/backend/shared/messaging/internal/natscap/adapter.go @@ -88,22 +88,22 @@ func (a *DurableAdapter) OpenConsumer(ctx context.Context, settings reliable.Dur return &natsSubscription{sub: sub}, nil } -func (a *DurableAdapter) EnsureProducerResources(_ context.Context, settings reliable.DurableSettings) error { +func (a *DurableAdapter) EnsureProducerResources(ctx context.Context, settings reliable.DurableSettings) error { if a == nil || a.broker == nil { return capability.Unavailable("durable admin") } - return ensureDurableStream(a.broker.Logger(), a.jetStream(), settings) + return ensureDurableStream(ctx, a.broker.Logger(), a.jetStream(), settings) } -func (a *DurableAdapter) EnsureConsumerResources(_ context.Context, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) error { +func (a *DurableAdapter) EnsureConsumerResources(ctx context.Context, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) error { if a == nil || a.broker == nil { return capability.Unavailable("durable admin") } js := a.jetStream() - if err := ensureDurableStream(a.broker.Logger(), js, settings); err != nil { + if err := ensureDurableStream(ctx, a.broker.Logger(), js, settings); err != nil { return err } - return ensureDurableConsumerConfig(a.broker.Logger(), js, settings, consumer) + return ensureDurableConsumerConfig(ctx, a.broker.Logger(), js, settings, consumer) } func (a *DurableAdapter) jetStream() nats.JetStreamContext { diff --git a/backend/shared/messaging/internal/natscap/admin.go b/backend/shared/messaging/internal/natscap/admin.go index c20502e..1c357dc 100644 --- a/backend/shared/messaging/internal/natscap/admin.go +++ b/backend/shared/messaging/internal/natscap/admin.go @@ -1,6 +1,7 @@ package natscap import ( + "context" "errors" "strings" "time" @@ -12,7 +13,7 @@ import ( "go.uber.org/zap" ) -func ensureDurableStream(logger *zap.Logger, js durableAdminContext, settings reliable.DurableSettings) error { +func ensureDurableStream(ctx context.Context, logger *zap.Logger, js durableAdminContext, settings reliable.DurableSettings) error { if js == nil { return capability.Unavailable("durable admin") } @@ -29,7 +30,7 @@ func ensureDurableStream(logger *zap.Logger, js durableAdminContext, settings re expected.Storage = nats.FileStorage } - info, err := js.StreamInfo(expected.Name) + info, err := js.StreamInfo(expected.Name, nats.Context(ctx)) if err != nil { if errors.Is(err, nats.ErrStreamNotFound) { if !settings.AllowsAutoProvision() { @@ -38,15 +39,15 @@ func ensureDurableStream(logger *zap.Logger, js durableAdminContext, settings re } return err } - _, err = js.AddStream(expected) + _, err = js.AddStream(expected, nats.Context(ctx)) return err } return err } current := info.Config - if err := validateStreamConfig(current, *expected, settings.IsProductionValidation()); err != nil { + if err := validateStreamConfig(current, *expected, settings.IsProductionValidation() || !settings.AllowsAutoProvision()); err != nil { if settings.AllowsAutoProvision() { - reconciled, reconcileErr := reconcileDurableStream(logger, js, current, *expected, settings) + reconciled, reconcileErr := reconcileDurableStream(ctx, logger, js, current, *expected, settings) if reconciled { return reconcileErr } @@ -58,13 +59,13 @@ func ensureDurableStream(logger *zap.Logger, js durableAdminContext, settings re } if settings.AllowsAutoProvision() && current.MaxAge != expected.MaxAge { current.MaxAge = expected.MaxAge - _, err = js.UpdateStream(¤t) + _, err = js.UpdateStream(¤t, nats.Context(ctx)) return err } return nil } -func reconcileDurableStream(logger *zap.Logger, js durableAdminContext, current nats.StreamConfig, expected nats.StreamConfig, settings reliable.DurableSettings) (bool, error) { +func reconcileDurableStream(ctx context.Context, logger *zap.Logger, js durableAdminContext, current nats.StreamConfig, expected nats.StreamConfig, settings reliable.DurableSettings) (bool, error) { if js == nil { return false, capability.Unavailable("durable admin") } @@ -90,14 +91,14 @@ func reconcileDurableStream(logger *zap.Logger, js durableAdminContext, current } current.Subjects = append([]string(nil), expected.Subjects...) current.MaxAge = expected.MaxAge - _, err := js.UpdateStream(¤t) + _, err := js.UpdateStream(¤t, nats.Context(ctx)) if err != nil { return true, err } return true, nil } -func ensureDurableConsumerConfig(logger *zap.Logger, js durableAdminContext, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) error { +func ensureDurableConsumerConfig(ctx context.Context, logger *zap.Logger, js durableAdminContext, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) error { if js == nil { return capability.Unavailable("durable admin") } @@ -117,7 +118,7 @@ func ensureDurableConsumerConfig(logger *zap.Logger, js durableAdminContext, set if consumer.InactiveThresholdSeconds > 0 { expected.InactiveThreshold = time.Duration(consumer.InactiveThresholdSeconds) * time.Second } - info, err := js.ConsumerInfo(settings.Stream.Name, consumer.Durable) + info, err := js.ConsumerInfo(settings.Stream.Name, consumer.Durable, nats.Context(ctx)) if err != nil { if errors.Is(err, nats.ErrConsumerNotFound) { if !settings.AllowsAutoProvision() { @@ -126,7 +127,7 @@ func ensureDurableConsumerConfig(logger *zap.Logger, js durableAdminContext, set } return err } - _, err = js.AddConsumer(settings.Stream.Name, expected) + _, err = js.AddConsumer(settings.Stream.Name, expected, nats.Context(ctx)) return err } return err @@ -142,14 +143,14 @@ func ensureDurableConsumerConfig(logger *zap.Logger, js durableAdminContext, set zap.Bool("recreate_required", true), ) } - if err := validateConsumerConfig(current, *expected, settings.IsProductionValidation()); err != nil { + if err := validateConsumerConfig(current, *expected, settings.IsProductionValidation() || !settings.AllowsAutoProvision()); err != nil { return err } if settings.AllowsAutoProvision() && (current.AckWait != expected.AckWait || current.MaxDeliver != expected.MaxDeliver || current.InactiveThreshold != expected.InactiveThreshold) { current.AckWait = expected.AckWait current.MaxDeliver = expected.MaxDeliver current.InactiveThreshold = expected.InactiveThreshold - _, err = js.UpdateConsumer(settings.Stream.Name, ¤t) + _, err = js.UpdateConsumer(settings.Stream.Name, ¤t, nats.Context(ctx)) return err } return nil @@ -202,7 +203,7 @@ func validateStreamConfig(current nats.StreamConfig, expected nats.StreamConfig, return merrors.DataConflict("durable stream replicas drift") } if strict && current.MaxAge != expected.MaxAge { - return nil + return merrors.DataConflict("durable stream retention age drift") } return nil } diff --git a/backend/shared/messaging/internal/natscap/subscription_test.go b/backend/shared/messaging/internal/natscap/subscription_test.go index f2592c8..c39901c 100644 --- a/backend/shared/messaging/internal/natscap/subscription_test.go +++ b/backend/shared/messaging/internal/natscap/subscription_test.go @@ -56,3 +56,15 @@ func TestConsumerDriftRejected(t *testing.T) { t.Fatal("invalid deliver policy accepted") } } + +func TestProductionStreamRetentionDriftRejected(t *testing.T) { + expected := nats.StreamConfig{Name: "NOTIFICATIONS", Subjects: []string{"notifications.email"}, Storage: nats.FileStorage, Retention: nats.LimitsPolicy, Replicas: 3, MaxAge: 48 * time.Hour} + current := expected + current.MaxAge = time.Hour + if err := validateStreamConfig(current, expected, true); err == nil { + t.Fatal("production retention drift accepted") + } + if err := validateStreamConfig(current, expected, false); err != nil { + t.Fatal("development retention cannot be reconciled", err) + } +} diff --git a/docs/implementation.md b/docs/implementation.md index e8f919f..d494b0a 100644 --- a/docs/implementation.md +++ b/docs/implementation.md @@ -480,3 +480,20 @@ in [PR #35](https://git.sendico.io/tech/sxchnge/pulls/35), which remains unmerge - Task: . PR: , branch `notifications-36`. Merge order: [#35](https://git.sendico.io/tech/sxchnge/pulls/35), then #37. + +### Notifications completion follow-up (#36 / PR #37) + +- Queue configuration now composes shared stream/provisioning settings. NATS + replication, retention and app/ops/hybrid provisioning are configurable; + production drift is rejected. Shared provisioning calls honor cancellation. +- Recovery expires queued and retrying notifications and clears rendered content + without requiring a working NATS connection, including future-scheduled retries. + Expiry queries have a dedicated index. Lost leases are logged with counts. +- Extended the real Mongo/NATS integration test through authenticated TLS gRPC + Send/Get, including caller isolation, idempotency and parameter propagation to + the email driver. Added queue configuration, cancellation and retention-drift + checks; gRPC now preserves canceled/deadline-exceeded outcomes. +- Validation: `make test`, `make lint-backend` and interface packaging passed in + an isolated clean checkout containing this contribution. Shared-workspace runs + initially failed because concurrent product/auth work had missing module + dependencies and unformatted files; that work was excluded and left untouched.