Complete notification expiry and configurable durable delivery (#36)

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,
+The queue uses `SXCHNGE_NOTIFICATIONS`, subject `sxchnge.notifications.email`,
-one replica), subject `sxchnge.notifications.email`, and durable consumer `email`.
+and durable consumer `email`. Configure `queue.stream.replicas` as 1, 3 or 5 to
-This initial setup is for a single-replica deployment. Configure production
+match the NATS cluster, and `queue.stream.max_age_hours` for retention (48 hours
-replication/provisioning policy in code before deploying to a clustered environment.
+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).
 

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)

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
 	}

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
+}

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")
+		}
+	}
+}

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) {
+func New(ctx context.Context, transport reliable.Transport, admin reliable.Admin, outbox Outbox, handler Handler, logger mlogger.Logger, config Config) (*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}}
+	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")

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

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"))
+}

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
+	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"
+	// Production ops mode validates existing resources without silently rewriting drift.
-	job, err := svc.Send(ctx, "product", request)
+	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())
+		result, e := client.Get(rpcCtx, &pb.GetRequest{Id: sent.Notification.Id})
-		if err != nil {
+		if e != nil {
-			t.Fatal(err)
+			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 != "<p>End to end &amp; escaped</p>" || delivered.Subject != "Hello" || delivered.Address != "user@example.com" {
+		t.Fatal("parameters did not reach driver", delivered)
+	}
 }

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) {

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

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")
+	}
+}

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 {

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(&current)
+		_, err = js.UpdateStream(&current, 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(&current)
+	_, err := js.UpdateStream(&current, 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, &current)
+		_, err = js.UpdateConsumer(settings.Stream.Name, &current, 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
 }

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)
+	}
+}

docs/implementation.md

@@ -480,3 +480,20 @@ in [PR #35](https://git.sendico.io/tech/sxchnge/pulls/35), which remains unmerge
 - Task: <https://git.sendico.io/tech/sxchnge/issues/36>.
   PR: <https://git.sendico.io/tech/sxchnge/pulls/37>, 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.