Merge pull request #987 from warmbly/fix/worker-send-deadline

feat: give worker sends a deadline that covers the SMTP budget and add ordered mailbox lanes on NATS

cmd/worker/main.go

@@ -223,7 +223,9 @@ func main() {
 		lanes = 1
 	}
 	log.Printf("Worker command concurrency: %d ordered mailbox lanes", lanes)
-	if err := eventbus.SubscribeKeyed(ctx, bus, []string{workerTopic}, "worker-"+workerID.String(), lanes, workerService.ResolveCommandKey, workerService.Receive); err != nil {
+	// A send runs past the bus's default handler budget, so the command subscription carries its own.
+	commandCtx := eventbus.WithHandlerTimeout(ctx, worker.CommandBudget)
+	if err := eventbus.SubscribeKeyed(commandCtx, bus, []string{workerTopic}, "worker-"+workerID.String(), lanes, workerService.ResolveCommandKey, workerService.Receive); err != nil {
 		log.Println("event bus subscribe ended:", err)
 	}
 

docs/content/docs/development/configuration.mdx

@@ -313,7 +313,7 @@ On `filesystem`, a remote worker writes blobs to its own disk rather than a volu
 | `KAFKA_SASL_USERNAME`, `KAFKA_SASL_PASSWORD` | Broker credentials | unset | yes |
 | `SCHEMA_REGISTRY_URL`, `SCHEMA_REGISTRY_KEY`, `SCHEMA_REGISTRY_SECRET` | Registry for the Avro codec | unset | yes |
 | `CODEC_PROVIDER` | `json` or `avro` | `json` under compose, `avro` for a bare binary | yes |
-| `EVENTBUS_HANDLER_TIMEOUT` | How long one handler may take before the delivery is abandoned | `30s` | yes |
+| `EVENTBUS_HANDLER_TIMEOUT` | How long one handler may take before the delivery is abandoned. A subscription that needs longer carries its own budget, never shorter than this one: a worker's command subscription runs under 195s, because a send is given 120s (the SMTP client's 90s conversation plus the checks before it) and then records its result. On NATS the ack wait is the subscription's budget plus 5s | `30s` | yes |
 | `PUBSUB_ENABLED` | `false` uses the Redis bridge for realtime fanout, `true` uses Google Pub/Sub | `false` | yes |
 | `GCP_PROJECT_ID` | Project when `PUBSUB_ENABLED=true` | unset | yes |
 
@@ -446,7 +446,7 @@ Keep the copies private. GeoLite2 may not be redistributed, which is also why no
 | `WORKER_BIND_IP` | Source address to bind outbound connections to, and the seed for a derived `WORKER_ID` | unset | yes |
 | `WORKER_PUBLIC_IP` | Public IPv4 reported by the worker, never replaced by the backend-observed address, which may be a reverse proxy. Otherwise the node discovers HTTPS egress IPv4 using `api.ipify.org` every ten minutes. Set this explicitly when mail and HTTPS have different egress routes | detected | yes |
 | `WARMBLY_WORKER_CAPACITY` | Assigned-mailbox planning target for this worker. This guides placement and is not a mailbox-provider send limit | `100` | yes |
-| `WORKER_COMMAND_CONCURRENCY` | Kafka worker command lanes, from `1` to `64`. Commands are grouped by the mailbox in the payload, including older task-keyed sends. Each mailbox stays in order; independent lanes run concurrently. NATS remains serial. Invalid values use the default | `8` | yes |
+| `WORKER_COMMAND_CONCURRENCY` | Worker command lanes, from `1` to `64`, on both Kafka and NATS. Commands are grouped by the mailbox in the payload, including older task-keyed sends. Each mailbox stays in order; independent lanes run concurrently, so one slow mail server holds only its own lane. On NATS a command waiting in a lane is kept from being redelivered while it waits. Invalid values use the default | `8` | yes |
 | `WARMBLY_NODE_REGION` | Free-form label for where this node egresses from, e.g. `eu-central`. Placement prefers a worker near where a mailbox's provider expects sign-ins; unset scores neutral. Written by the join script from `--region` | unset | yes |
 | `ENCRYPTED_KEYS_BACKEND_URL` | Backend base the worker fetches organization keys from | unset | yes |
 | `ENCRYPTED_KEYS_WORKER_TOKEN` | The worker's copy of `INTERNAL_API_TOKEN` | unset | yes |

docs/content/docs/development/events.mdx

@@ -91,6 +91,8 @@ The consumer registers one handler per type (`internal/app/consumer/events.go`);
 
 Every `SEND_EMAIL` is answered with exactly one per-task result: `EMAIL_SENT` (the provider accepted the message; the consumer records the wire Message-ID on the task) or `EMAIL_FAILED` (a `SendEmailResult` carrying the error code and message). The control plane stamps a campaign step sent when it hands the send to the worker, so `EMAIL_FAILED` is what walks that back: the consumer marks the task failed, clears the step's `sent_at` and counts the attempt on `campaign_contact_progress`, gives the send back to the campaign's daily counters, writes the failure to the campaign activity log, and reopens a campaign that completed while the send was in flight. After `CampaignSendMaxAttempts` (5) the lead is marked failed and routing drops it; a `RECIPIENT_REJECTED` result (refused at RCPT) skips the retries and is ingested as a bounce instead. Account-level conditions (`EMAIL_AUTH_ERROR`, `EMAIL_DISABLED`, `EMAIL_RATE_LIMITED`, `EMAIL_SERVER_ERROR`) are raised in addition to, never instead of, the per-task result; they carry an `EmailErrorEvent` and act on the mailbox. A worker that does not hold the mailbox yet leaves the send for a few redeliveries before reporting it failed, and the backend never publishes a send to a worker that is not heartbeating.
 
+A worker gives a `SEND_EMAIL` 120 seconds, enough for the SMTP client's own 90-second conversation with a slow server plus the dispatch checks and body fetch before it; every other command keeps 30 seconds. Its command subscription runs under 195 seconds, which also covers recording the result, so the bus does not hand a send out again while the first delivery is still finishing. Commands run on ordered per-mailbox lanes (`WORKER_COMMAND_CONCURRENCY`) on both buses, so a stalled mail server holds only its own mailbox's commands.
+
 `UPDATE_FOLDER` reports the canonical folder a provider now has a message in, without the rest of the message. Gmail sends it when a history record adds or removes `INBOX`, `TRASH`, `SPAM`, `SENT` or `DRAFT`, with the folder recomputed from the message's labels, and from a reconciliation that runs every six hours and once per mailbox load: the worker lists the Gmail inbox and asks the backend (`GET /api/v1/internal/sync/provider-folder-messages`) which rows it believes are in inbox, archive, spam or trash, then reports the ones Gmail has elsewhere. The consumer resolves it like `UPDATE_EMAIL`: against the folder the provider last reported, so a message filed in Warmbly stays filed unless the provider itself moved it. The worker also sends it, with `relayed` set, for every message a `MESSAGE_FOLDER` left in its destination, plus the handles the move changed (the new Graph id in `provider_id`, or the IMAP `folder_path`, `mailbox` and `uid`). A relayed one writes `provider_folder` and those handles and never the folder, so the sync's own report of the same placement afterwards reads as no move, and a filing undone meanwhile stays undone.
 
 `SYNC_STATE` is the worker's relay of a mailbox's sync state (backfill progress and cursor, fair-use throttle, last-synced time). It carries the full state rather than a delta, so a lost event is repaired by the next one; the consumer writes it to `email_sync_state` and the backend hands it back inside `ADD_EMAIL` on the next load, which is what lets a replaced worker resume an import instead of restarting it.
@@ -145,7 +147,7 @@ EVENTBUS_PROVIDER=nats            # nats | kafka (kafka needs the kafka build ta
 NATS_URL=nats://nats:4222
 NATS_STREAM_NAME=warmbly          # optional, default warmbly
 NATS_SUBJECT_PREFIX=warmbly       # optional, default warmbly
-EVENTBUS_HANDLER_TIMEOUT=30s      # optional, per-message handler timeout
+EVENTBUS_HANDLER_TIMEOUT=30s      # optional, per-message handler timeout (worker commands run under 195s)
 
 # Codec
 CODEC_PROVIDER=json               # json | avro (avro needs the kafka build tag)

docs/content/docs/development/send-safety.mdx

@@ -100,7 +100,7 @@ The PATCH atomically queues an immutable audited result; it does **not** immedia
 
 Campaign routing, placement, day plans and estimates share durable mailbox send admission. Recovery-held or currently cooling mailboxes contribute no immediately available capacity and do not trap a lead when another healthy sender is available. Plans identify send recovery and provider cooldown separately. When the entire pool is unavailable, the campaign defers with a bounded recheck rather than completing or repeatedly dispatching refused sends. An admission read failure also closes the affected mailbox for that pass.
 
-After provider execution, workers record and publish the result on an independent 45-second context with at most four attempts, each bounded to 10 seconds. Failed reports retain the exact outcome in memory for command redelivery without another provider send. Persisted results survive worker replacement and lost broker events: the consumer replays them through the same idempotent reconciliation. A worker crash before durable recording can still leave an unknown outcome requiring investigation.
+Provider execution gets 120 seconds, which covers the SMTP client's own 90-second budget for a slow server's verdict after the message, so a send is not cut off waiting for an answer the server was about to give. After provider execution, workers record and publish the result on an independent 45-second context with at most four attempts, each bounded to 10 seconds. Failed reports retain the exact outcome in memory for command redelivery without another provider send. Persisted results survive worker replacement and lost broker events: the consumer replays them through the same idempotent reconciliation. A worker crash before durable recording can still leave an unknown outcome requiring investigation.
 
 Already-failed tasks with no outbound reservation or execution authorization cannot create unknown provider holds. The recorded-result sweep also reconciles existing invalid holds for such tasks. Real reserved or started unknown sends remain held; upgrades never infer failure from elapsed time. Deploy the backend/consumer (including the additive migration) before updated workers. No new configuration is required; existing genuine unknown holds need definitive worker evidence or the explicit operator confirmation above.
 

internal/app/instanceconfig/entries.go

@@ -690,7 +690,7 @@ var table = []Entry{
 	},
 	{
 		Key: "EVENTBUS_HANDLER_TIMEOUT", Group: GroupEventBus, RuntimeChangeable: ChangeBootOnly,
-		Effect:     "How long one consumer handler may run before the message is redelivered.",
+		Effect:     "How long one consumer handler may run before the message is redelivered. Worker commands carry a longer budget of their own.",
 		DocsAnchor: docsEventBus,
 		Resolve:    envValue("EVENTBUS_HANDLER_TIMEOUT"),
 	},

internal/app/worker/receive.go

@@ -4,10 +4,29 @@ import (
 	"context"
 	"time"
 
+	"github.com/warmbly/warmbly/internal/client/smtpimap/smtp"
 	"github.com/warmbly/warmbly/internal/infrastructure/eventbus"
 	"github.com/warmbly/warmbly/internal/models"
 )
 
+const (
+	// eventTimeout bounds every worker command except a send.
+	eventTimeout = 30 * time.Second
+	// sendEventTimeout covers the SMTP client's whole conversation plus the
+	// dispatch checks and body fetch that run before it.
+	sendEventTimeout = smtp.SendTimeout + 30*time.Second
+	// sendResultReportTimeout is the detached budget a send's outcome is recorded under.
+	sendResultReportTimeout = 45 * time.Second
+	// publishTimeout bounds the mailbox error event a failed send raises (the bus's default publish bound).
+	publishTimeout = 30 * time.Second
+)
+
+// CommandBudget is the longest one worker command can hold its delivery: a
+// send's deadline, its result report and the mailbox error it may raise. The
+// command subscription runs under it so the bus does not hand a command out
+// again while the first delivery is still working.
+const CommandBudget = sendEventTimeout + sendResultReportTimeout + publishTimeout
+
 type deliveryKey struct{}
 
 // delivery is what a handler may know about the bus message it is processing.
@@ -23,6 +42,14 @@ func deliveryOf(ctx context.Context) delivery {
 	return d
 }
 
+// eventDeadline is how long one command of the given type may run.
+func eventDeadline(t models.WorkerEventType) time.Duration {
+	if t == models.WorkerEventTypeSendEmail {
+		return sendEventTimeout
+	}
+	return eventTimeout
+}
+
 // Receive is the eventbus.Handler that drives the worker's event loop. It
 // decodes the wire payload via the injected codec.Codec and dispatches to
 // HandleEvent.
@@ -32,7 +59,7 @@ func (w *WorkerService) Receive(ctx context.Context, msg eventbus.Message) error
 		return err
 	}
 
-	hctx, cancel := context.WithTimeout(context.WithValue(ctx, deliveryKey{}, delivery{attempt: msg.Attempt, redelivers: msg.Redelivers}), 30*time.Second)
+	hctx, cancel := context.WithTimeout(context.WithValue(ctx, deliveryKey{}, delivery{attempt: msg.Attempt, redelivers: msg.Redelivers}), eventDeadline(event.Type))
 	defer cancel()
 	return w.HandleEvent(hctx, &event)
 }

internal/app/worker/receive_test.go

@@ -0,0 +1,56 @@
+package worker
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/warmbly/warmbly/internal/client/smtpimap/smtp"
+	"github.com/warmbly/warmbly/internal/infrastructure/codec"
+	"github.com/warmbly/warmbly/internal/infrastructure/eventbus"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+// A send's deadline must outlast the SMTP client's own budget, or a slow
+// server's verdict is cut off after DATA and the result becomes unknown.
+func TestReceiveGivesASendTheSMTPBudget(t *testing.T) {
+	w := &WorkerService{Codec: codec.NewJSON()}
+	left := map[models.WorkerEventType]time.Duration{}
+	record := func(t models.WorkerEventType) func(context.Context, any) error {
+		return func(ctx context.Context, _ any) error {
+			d, ok := ctx.Deadline()
+			if !ok {
+				left[t] = -1
+				return nil
+			}
+			left[t] = time.Until(d)
+			return nil
+		}
+	}
+	w.eventHandlers = map[models.WorkerEventType]func(context.Context, any) error{
+		models.WorkerEventTypeSendEmail: record(models.WorkerEventTypeSendEmail),
+		models.WorkerEventTypeAddEmail:  record(models.WorkerEventTypeAddEmail),
+	}
+	// The bus hands the handler the subscription's budget; Receive narrows it per event.
+	ctx, cancel := context.WithTimeout(t.Context(), CommandBudget)
+	defer cancel()
+	for typ := range w.eventHandlers {
+		payload, err := w.Codec.Serialize(ctx, "commands", &models.WorkerEvent{Type: typ, Body: map[string]any{}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := w.Receive(ctx, eventbus.Message{Payload: payload}); err != nil {
+			t.Fatalf("%s: %v", typ, err)
+		}
+	}
+
+	if send := left[models.WorkerEventTypeSendEmail]; send <= smtp.SendTimeout {
+		t.Fatalf("send deadline %s does not cover the SMTP budget %s", send, smtp.SendTimeout)
+	}
+	if other := left[models.WorkerEventTypeAddEmail]; other <= 0 || other > eventTimeout {
+		t.Fatalf("non-send deadline %s, want at most %s", other, eventTimeout)
+	}
+	if CommandBudget < sendEventTimeout+sendResultReportTimeout {
+		t.Fatalf("command budget %s ends before a send and its result report can", CommandBudget)
+	}
+}

internal/app/worker/send_result_report.go

@@ -38,7 +38,7 @@ func (w *WorkerService) replayPendingSend(ctx context.Context, command models.Se
 
 // Provider execution spends the handler budget; recording gets its own bounded budget.
 func (w *WorkerService) reportSendResult(parent context.Context, p pendingSendResult) error {
-	ctx, cancel := context.WithTimeout(context.WithoutCancel(parent), 45*time.Second)
+	ctx, cancel := context.WithTimeout(context.WithoutCancel(parent), sendResultReportTimeout)
 	defer cancel()
 	var err error
 	for attempt := range 4 {

internal/client/smtpimap/smtp/client.go

@@ -26,11 +26,11 @@ import (
 	"golang.org/x/oauth2"
 )
 
-// sendTimeout bounds one whole SMTP conversation: dial, greeting, AUTH, the
+// SendTimeout bounds one whole SMTP conversation: dial, greeting, AUTH, the
 // recipients, the message and the server's verdict. It is generous because a
 // large message to a slow relay is legitimate, and it exists so a peer that
 // stops answering cannot hold a send goroutine forever.
-const sendTimeout = 90 * time.Second
+const SendTimeout = 90 * time.Second
 
 // errNoSTARTTLS distinguishes a missing upgrade from a network failure.
 var errNoSTARTTLS = errors.New("the server does not offer STARTTLS")
@@ -332,7 +332,7 @@ func (c *Client) sendRaw(ctx context.Context, from string, to []string, data []b
 	// without closing the connection parked the send goroutine forever: the
 	// greeting, AUTH, each RCPT and the wait for the server's verdict at the
 	// end of DATA all block with nothing to fail them.
-	ctx, cancel := context.WithTimeout(ctx, sendTimeout)
+	ctx, cancel := context.WithTimeout(ctx, SendTimeout)
 	defer cancel()
 
 	resolved := models.ResolveSMTPSecurity(security, port)

internal/infrastructure/eventbus/bus.go

@@ -57,7 +57,8 @@ type EventBus interface {
 	// Subscribe registers a Handler against one or more topics under a named
 	// consumer group. The call blocks until the context is cancelled or a
 	// fatal error occurs. The handler is invoked one message at a time per
-	// subscription. A handler that returns nil signals successful processing
+	// subscription, under a deadline of EVENTBUS_HANDLER_TIMEOUT or the longer
+	// one set by WithHandlerTimeout. A handler that returns nil signals successful processing
 	// and the message is acked; a non-nil error leaves the message for
 	// redelivery according to the backend's retry policy.
 	Subscribe(ctx context.Context, topics []string, group string, handler Handler) error
@@ -184,3 +185,21 @@ func handlerTimeout() time.Duration {
 	}
 	return 30 * time.Second
 }
+
+type handlerTimeoutKey struct{}
+
+// WithHandlerTimeout raises the per-message budget of subscriptions opened with
+// ctx, for handlers that legitimately run longer. It never lowers
+// EVENTBUS_HANDLER_TIMEOUT.
+func WithHandlerTimeout(ctx context.Context, d time.Duration) context.Context {
+	return context.WithValue(ctx, handlerTimeoutKey{}, d)
+}
+
+// subscriptionTimeout is the per-message budget of a subscription opened with ctx.
+func subscriptionTimeout(ctx context.Context) time.Duration {
+	d := handlerTimeout()
+	if v, ok := ctx.Value(handlerTimeoutKey{}).(time.Duration); ok && v > d {
+		return v
+	}
+	return d
+}

internal/infrastructure/eventbus/kafka.go

@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	ckf "github.com/confluentinc/confluent-kafka-go/v2/kafka"
 	"github.com/rs/zerolog/log"
@@ -146,6 +147,7 @@ func (b *KafkaBus) subscribeOnce(ctx context.Context, topics []string, group str
 	// background: a synchronous commit per message cost a broker round trip each.
 	cc.Set("enable.auto.commit", true)
 	cc.Set("enable.auto.offset.store", false)
+	cc.Set("max.poll.interval.ms", int(maxPollInterval(subscriptionTimeout(ctx)).Milliseconds()))
 
 	cons, err := cc.Connect()
 	if err != nil {
@@ -191,6 +193,7 @@ func (b *KafkaBus) subscribeOnce(ctx context.Context, topics []string, group str
 }
 
 func kafkaHandler(ctx context.Context, handler Handler) func(*ckf.Message) error {
+	budget := subscriptionTimeout(ctx)
 	var mu sync.Mutex
 	attempts := map[*ckf.Message]int{}
 	return func(msg *ckf.Message) error {
@@ -201,7 +204,7 @@ func kafkaHandler(ctx context.Context, handler Handler) func(*ckf.Message) error
 		envelope := kafkaEnvelope(msg)
 		envelope.Attempt = attempt
 		// A message already being handled finishes after a shutdown signal; only reading the next one stops.
-		hctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), handlerTimeout())
+		hctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget)
 		defer cancel()
 		if err := invokeHandler(hctx, handler, envelope); err != nil {
 			log.Error().Err(err).Str("topic", envelope.Topic).Msg("eventbus kafka handler error")
@@ -214,6 +217,12 @@ func kafkaHandler(ctx context.Context, handler Handler) func(*ckf.Message) error
 	}
 }
 
+// maxPollInterval leaves a serial consumer room for a handler that runs its
+// whole budget and is retried once, keeping librdkafka's 5 minute default as the floor.
+func maxPollInterval(budget time.Duration) time.Duration {
+	return max(5*time.Minute, 2*budget+30*time.Second)
+}
+
 func kafkaEnvelope(msg *ckf.Message) Message {
 	topic := ""
 	if msg.TopicPartition.Topic != nil {

internal/infrastructure/eventbus/kafka_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	ckf "github.com/confluentinc/confluent-kafka-go/v2/kafka"
 	"github.com/warmbly/warmbly/internal/infrastructure/kafka"
@@ -187,3 +188,13 @@ func TestKafkaTopicHelpersRoundTrip(t *testing.T) {
 		t.Fatalf("worker-events topic translation drift: %q", Subject(kafka.TopicWorkerEvents))
 	}
 }
+
+// A serial consumer must not be evicted from its group while one handler runs its whole budget.
+func TestMaxPollIntervalCoversTheHandlerBudget(t *testing.T) {
+	if got := maxPollInterval(30 * time.Second); got != 5*time.Minute {
+		t.Fatalf("default budget poll interval = %s, want librdkafka's 5m", got)
+	}
+	if got := maxPollInterval(195 * time.Second); got < 2*195*time.Second {
+		t.Fatalf("poll interval %s does not cover a long handler and its retry", got)
+	}
+}

internal/infrastructure/eventbus/nats.go

@@ -14,6 +14,10 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
+// natsAckGrace keeps the ack wait past the handler budget, so a message is not
+// handed out again while its handler can still be running.
+var natsAckGrace = 5 * time.Second
+
 // NATSBus is the EventBus backed by NATS JetStream. JetStream (not core NATS)
 // is required because the rest of the system assumes durable, replayable
 // delivery: a worker that goes offline must resume from its last ack.
@@ -223,6 +227,12 @@ func (b *NATSBus) Publish(ctx context.Context, topic, key string, payload []byte
 }
 
 func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string, handler Handler) error {
+	return b.SubscribeKeyed(ctx, topics, group, 1, nil, handler)
+}
+
+// SubscribeKeyed hands messages with the same resolved key to one ordered lane
+// and runs independent lanes concurrently, as the Kafka backend does.
+func (b *NATSBus) SubscribeKeyed(ctx context.Context, topics []string, group string, lanes int, key KeyResolver, handler Handler) error {
 	if len(topics) == 0 {
 		return errors.New("eventbus nats: at least one topic required")
 	}
@@ -232,6 +242,9 @@ func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string,
 	if handler == nil {
 		return errors.New("eventbus nats: handler required")
 	}
+	if lanes < 1 || lanes > 64 || (lanes > 1 && key == nil) {
+		return errors.New("eventbus nats: keyed subscription requires 1-64 lanes and a key resolver")
+	}
 	if err := b.ensureStream(ctx); err != nil {
 		return err
 	}
@@ -246,44 +259,25 @@ func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string,
 		return fmt.Errorf("eventbus nats: lookup stream: %w", err)
 	}
 
+	budget := subscriptionTimeout(ctx)
+	ackWait := budget + natsAckGrace
 	cons, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
 		Durable:        b.durable(group, topics),
 		AckPolicy:      jetstream.AckExplicitPolicy,
 		DeliverPolicy:  jetstream.DeliverAllPolicy,
 		FilterSubjects: subjects,
 		MaxDeliver:     10,
-		AckWait:        handlerTimeout() + 5*time.Second,
+		AckWait:        ackWait,
 	})
 	if err != nil {
 		return fmt.Errorf("eventbus nats: create consumer: %w", err)
 	}
 
-	cc, err := cons.Consume(func(m jetstream.Msg) {
+	deliver := func(m jetstream.Msg) {
 		// A message already being handled finishes after a shutdown signal; only reading the next one stops.
-		hctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), handlerTimeout())
+		hctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget)
 		defer cancel()
-		key := m.Headers().Get("Warmbly-Key")
-		if key == "" {
-			key = m.Headers().Get(nats.MsgIdHdr)
-		}
-		// Strip the prefix when reporting topic back to handlers so they see
-		// the same string they published with. Note: NATS subjects use "."
-		// while Kafka topics use ":" — handlers may receive either form
-		// depending on which backend is wired in. New code should not rely
-		// on the exact separator and should use Subject() if it needs to
-		// compare against a Kafka-style topic name.
-		topic := strings.TrimPrefix(m.Subject(), b.prefix+".")
-		attempt := 1
-		if meta, merr := m.Metadata(); merr == nil && meta.NumDelivered > 0 {
-			attempt = int(meta.NumDelivered)
-		}
-		if err := invokeHandler(hctx, handler, Message{
-			Topic:      topic,
-			Key:        key,
-			Payload:    m.Data(),
-			Attempt:    attempt,
-			Redelivers: true,
-		}); err != nil {
+		if err := invokeHandler(hctx, handler, b.envelope(m)); err != nil {
 			log.Error().Err(err).Str("subject", m.Subject()).Msg("eventbus nats handler error")
 			// Nak with a short delay so transient errors don't hot-loop.
 			if nakErr := m.NakWithDelay(time.Second); nakErr != nil {
@@ -294,16 +288,47 @@ func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string,
 		if ackErr := m.Ack(); ackErr != nil {
 			log.Warn().Err(ackErr).Msg("eventbus nats ack failed")
 		}
-	})
+	}
+
+	var pool *natsLanes
+	callback := deliver
+	var opts []jetstream.PullConsumeOpt
+	if lanes > 1 {
+		pool = newNATSLanes(lanes, ackWait, deliver)
+		callback = func(m jetstream.Msg) {
+			laneKey, err := resolveKey(ctx, key, b.envelope(m))
+			if err != nil {
+				log.Error().Err(err).Str("subject", m.Subject()).Msg("eventbus nats key resolution failed")
+				if nakErr := m.NakWithDelay(time.Second); nakErr != nil {
+					log.Warn().Err(nakErr).Msg("eventbus nats nak failed")
+				}
+				return
+			}
+			pool.dispatch(m, laneKey)
+		}
+		// Fetch again only after dispatch tracks this delivery, even when a lane is full.
+		opts = append(opts, jetstream.PullMaxMessages(1))
+	}
+
+	cc, err := cons.Consume(callback, opts...)
 	if err != nil {
+		if pool != nil {
+			pool.stop()
+		}
 		return fmt.Errorf("eventbus nats: consume: %w", err)
 	}
 
 	b.mu.Lock()
 	if b.closed {
 		b.mu.Unlock()
+		if pool != nil {
+			pool.halt()
+		}
 		cc.Stop()
 		<-cc.Closed()
+		if pool != nil {
+			pool.stop()
+		}
 		return ErrBusClosed
 	}
 	b.subscribers = append(b.subscribers, cc)
@@ -311,12 +336,38 @@ func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string,
 
 	// Block until ctx is cancelled, mirroring KafkaBus.Subscribe semantics.
 	<-ctx.Done()
+	if pool != nil {
+		pool.halt()
+	}
 	cc.Stop()
 	// Closed fires once the callback in progress has returned, so shutdown waits for it.
 	<-cc.Closed()
+	if pool != nil {
+		pool.stop()
+	}
 	return ctx.Err()
 }
 
+// envelope reports a JetStream message in the bus-neutral form handlers read.
+func (b *NATSBus) envelope(m jetstream.Msg) Message {
+	key := m.Headers().Get("Warmbly-Key")
+	if key == "" {
+		key = m.Headers().Get(nats.MsgIdHdr)
+	}
+	attempt := 1
+	if meta, err := m.Metadata(); err == nil && meta.NumDelivered > 0 {
+		attempt = int(meta.NumDelivered)
+	}
+	return Message{
+		// Handlers see the topic they published to, without the stream prefix.
+		Topic:      strings.TrimPrefix(m.Subject(), b.prefix+"."),
+		Key:        key,
+		Payload:    m.Data(),
+		Attempt:    attempt,
+		Redelivers: true,
+	}
+}
+
 func (b *NATSBus) Close() error {
 	b.mu.Lock()
 	if b.closed {
@@ -352,7 +403,10 @@ func natsURLFromEnv() string {
 }
 
 // Compile-time interface check.
-var _ EventBus = (*NATSBus)(nil)
+var (
+	_ EventBus        = (*NATSBus)(nil)
+	_ KeyedSubscriber = (*NATSBus)(nil)
+)
 
 // maxBytesOrUnlimited maps an unset ceiling onto JetStream's own spelling for
 // it. Zero in a StreamConfig means zero bytes, not unlimited.

internal/infrastructure/eventbus/nats_lanes.go

@@ -0,0 +1,144 @@
+package eventbus
+
+import (
+	"hash/fnv"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/rs/zerolog/log"
+)
+
+const natsLaneBuffer = 32
+
+// natsLanes runs a keyed subscription on ordered lanes. JetStream times a
+// delivery from when it was handed out, so every message the lanes hold,
+// queued or running, is kept alive with an in-progress heartbeat.
+type natsLanes struct {
+	queues   []chan jetstream.Msg
+	deliver  func(jetstream.Msg)
+	halted   chan struct{}
+	haltOnce sync.Once
+	done     chan struct{}
+	wg       sync.WaitGroup
+
+	mu   sync.Mutex
+	held map[jetstream.Msg]struct{}
+}
+
+func newNATSLanes(lanes int, ackWait time.Duration, deliver func(jetstream.Msg)) *natsLanes {
+	p := &natsLanes{
+		queues:  make([]chan jetstream.Msg, lanes),
+		deliver: deliver,
+		halted:  make(chan struct{}),
+		done:    make(chan struct{}),
+		held:    map[jetstream.Msg]struct{}{},
+	}
+	for i := range p.queues {
+		p.queues[i] = make(chan jetstream.Msg, natsLaneBuffer)
+		p.wg.Add(1)
+		go p.run(p.queues[i])
+	}
+	go p.heartbeat(max(ackWait/3, 100*time.Millisecond))
+	return p
+}
+
+func natsLane(key string, lanes int) int {
+	h := fnv.New32a()
+	_, _ = h.Write([]byte(key))
+	return int(h.Sum32() % uint32(lanes))
+}
+
+// dispatch queues m behind every earlier message with the same key.
+func (p *natsLanes) dispatch(m jetstream.Msg, key string) {
+	p.hold(m)
+	if p.isHalted() {
+		p.abandon(m)
+		return
+	}
+	select {
+	case p.queues[natsLane(key, len(p.queues))] <- m:
+	case <-p.halted:
+		p.abandon(m)
+	}
+}
+
+func (p *natsLanes) run(queue <-chan jetstream.Msg) {
+	defer p.wg.Done()
+	for m := range queue {
+		if p.isHalted() {
+			p.abandon(m)
+			continue
+		}
+		p.deliver(m)
+		p.release(m)
+	}
+}
+
+func (p *natsLanes) hold(m jetstream.Msg) {
+	p.mu.Lock()
+	p.held[m] = struct{}{}
+	p.mu.Unlock()
+}
+
+func (p *natsLanes) release(m jetstream.Msg) {
+	p.mu.Lock()
+	delete(p.held, m)
+	p.mu.Unlock()
+}
+
+// abandon hands back a message that will not be handled here, so it is
+// redelivered without waiting out its ack wait.
+func (p *natsLanes) abandon(m jetstream.Msg) {
+	p.release(m)
+	if err := m.Nak(); err != nil {
+		log.Warn().Err(err).Msg("eventbus nats nak failed")
+	}
+}
+
+func (p *natsLanes) heartbeat(every time.Duration) {
+	ticker := time.NewTicker(every)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-p.done:
+			return
+		case <-ticker.C:
+		}
+		p.mu.Lock()
+		msgs := make([]jetstream.Msg, 0, len(p.held))
+		for m := range p.held {
+			msgs = append(msgs, m)
+		}
+		p.mu.Unlock()
+		for _, m := range msgs {
+			// A message settled since the snapshot refuses this, which is fine.
+			if err := m.InProgress(); err != nil {
+				log.Debug().Err(err).Msg("eventbus nats in-progress heartbeat failed")
+			}
+		}
+	}
+}
+
+func (p *natsLanes) isHalted() bool {
+	select {
+	case <-p.halted:
+		return true
+	default:
+		return false
+	}
+}
+
+// halt stops handling anything not already running; queued messages are handed back.
+func (p *natsLanes) halt() {
+	p.haltOnce.Do(func() { close(p.halted) })
+}
+
+// stop waits for the running handlers once no more messages can be dispatched.
+func (p *natsLanes) stop() {
+	for _, q := range p.queues {
+		close(q)
+	}
+	p.wg.Wait()
+	close(p.done)
+}

internal/infrastructure/eventbus/nats_lanes_test.go

@@ -0,0 +1,251 @@
+package eventbus
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestSubscriptionTimeout_RaisesNeverLowers(t *testing.T) {
+	t.Setenv("EVENTBUS_HANDLER_TIMEOUT", "30s")
+	base := context.Background()
+	if got := subscriptionTimeout(base); got != 30*time.Second {
+		t.Fatalf("default budget = %s, want 30s", got)
+	}
+	if got := subscriptionTimeout(WithHandlerTimeout(base, 195*time.Second)); got != 195*time.Second {
+		t.Fatalf("raised budget = %s, want 195s", got)
+	}
+	if got := subscriptionTimeout(WithHandlerTimeout(base, 5*time.Second)); got != 30*time.Second {
+		t.Fatalf("a shorter budget must not lower EVENTBUS_HANDLER_TIMEOUT, got %s", got)
+	}
+}
+
+// subscribeInBackground runs SubscribeKeyed until the test ends.
+func subscribeInBackground(t *testing.T, ctx context.Context, bus *NATSBus, topic, group string, lanes int, key KeyResolver, h Handler) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(ctx)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		_ = bus.SubscribeKeyed(ctx, []string{topic}, group, lanes, key, h)
+	}()
+	t.Cleanup(func() {
+		cancel()
+		<-done
+	})
+}
+
+func waitForConsumer(t *testing.T, bus *NATSBus, durable string) time.Duration {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if s, err := bus.js.Stream(t.Context(), bus.stream); err == nil {
+			if c, err := s.Consumer(t.Context(), durable); err == nil {
+				if info, err := c.Info(t.Context()); err == nil {
+					return info.Config.AckWait
+				}
+			}
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatalf("consumer %s never appeared", durable)
+	return 0
+}
+
+func TestNATSBus_HandlerTimeoutSetsBudgetAndAckWait(t *testing.T) {
+	t.Setenv("EVENTBUS_HANDLER_TIMEOUT", "30s")
+	bus := newTestNATSBus(t)
+	topic := "w.budget"
+
+	deadlines := make(chan time.Duration, 1)
+	ctx := WithHandlerTimeout(t.Context(), 3*time.Minute)
+	subscribeInBackground(t, ctx, bus, topic, "budget", 1, nil, func(ctx context.Context, _ Message) error {
+		d, _ := ctx.Deadline()
+		deadlines <- time.Until(d)
+		return nil
+	})
+
+	if got, want := waitForConsumer(t, bus, bus.durable("budget", []string{topic})), 3*time.Minute+natsAckGrace; got != want {
+		t.Fatalf("ack wait = %s, want %s", got, want)
+	}
+	if err := bus.Publish(t.Context(), topic, "k", []byte("x")); err != nil {
+		t.Fatalf("publish: %v", err)
+	}
+	select {
+	case left := <-deadlines:
+		if left < 2*time.Minute || left > 3*time.Minute {
+			t.Fatalf("handler deadline %s away, want about 3m", left)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("handler never ran")
+	}
+}
+
+func payloadKey(_ context.Context, msg Message) (string, error) {
+	return string(msg.Payload[:1]), nil
+}
+
+func TestNATSBus_KeyedLanesKeepKeyOrderAndRunKeysConcurrently(t *testing.T) {
+	bus := newTestNATSBus(t)
+	topic := "w.lanes"
+
+	releaseA := make(chan struct{})
+	var mu sync.Mutex
+	var order []string
+	done := make(chan struct{})
+	subscribeInBackground(t, t.Context(), bus, topic, "lanes", 4, payloadKey, func(_ context.Context, msg Message) error {
+		p := string(msg.Payload)
+		if p == "a1" {
+			// Held until another key is handled, which only a concurrent lane can do.
+			select {
+			case <-releaseA:
+			case <-time.After(5 * time.Second):
+			}
+		}
+		mu.Lock()
+		order = append(order, p)
+		n := len(order)
+		mu.Unlock()
+		if p == "b1" {
+			close(releaseA)
+		}
+		if n == 4 {
+			close(done)
+		}
+		return nil
+	})
+	// a and b must not share a lane for the concurrency half to mean anything.
+	if natsLane("a", 4) == natsLane("b", 4) {
+		t.Fatal("test keys hash to the same lane")
+	}
+	waitForConsumer(t, bus, bus.durable("lanes", []string{topic}))
+	for _, p := range []string{"a1", "a2", "a3", "b1"} {
+		if err := bus.Publish(t.Context(), topic, "", []byte(p)); err != nil {
+			t.Fatalf("publish %s: %v", p, err)
+		}
+	}
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("not every message was handled")
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if order[0] != "b1" {
+		t.Fatalf("b1 waited behind another key's lane: %v", order)
+	}
+	if got := order[1:]; got[0] != "a1" || got[1] != "a2" || got[2] != "a3" {
+		t.Fatalf("one key's messages ran out of order: %v", order)
+	}
+}
+
+// A message queued behind a slow one on its lane outlives its ack wait; the
+// in-progress heartbeat must keep JetStream from handing it out again.
+func TestNATSBus_KeyedLanesKeepQueuedMessagesAlive(t *testing.T) {
+	grace := natsAckGrace
+	natsAckGrace = 200 * time.Millisecond
+	t.Cleanup(func() { natsAckGrace = grace })
+	t.Setenv("EVENTBUS_HANDLER_TIMEOUT", "300ms")
+
+	bus := newTestNATSBus(t)
+	topic := "w.heartbeat"
+
+	var mu sync.Mutex
+	seen := map[string][]int{}
+	subscribeInBackground(t, t.Context(), bus, topic, "heartbeat", 2, payloadKey, func(_ context.Context, msg Message) error {
+		mu.Lock()
+		seen[string(msg.Payload)] = append(seen[string(msg.Payload)], msg.Attempt)
+		mu.Unlock()
+		if string(msg.Payload) == "a1" {
+			// Well past the 500ms ack wait, ignoring the handler deadline.
+			time.Sleep(1500 * time.Millisecond)
+		}
+		return nil
+	})
+	waitForConsumer(t, bus, bus.durable("heartbeat", []string{topic}))
+	for _, p := range []string{"a1", "a2"} {
+		if err := bus.Publish(t.Context(), topic, "", []byte(p)); err != nil {
+			t.Fatalf("publish %s: %v", p, err)
+		}
+	}
+	time.Sleep(3 * time.Second)
+
+	mu.Lock()
+	defer mu.Unlock()
+	for _, p := range []string{"a1", "a2"} {
+		if got := seen[p]; len(got) != 1 || got[0] != 1 {
+			t.Fatalf("%s handled with attempts %v, want exactly one first delivery", p, got)
+		}
+	}
+}
+
+func TestNATSBus_SubscribeKeyedValidatesLanes(t *testing.T) {
+	bus := newTestNATSBus(t)
+	noop := func(context.Context, Message) error { return nil }
+	if err := bus.SubscribeKeyed(t.Context(), []string{"w.x"}, "g", 0, payloadKey, noop); err == nil {
+		t.Fatal("zero lanes accepted")
+	}
+	if err := bus.SubscribeKeyed(t.Context(), []string{"w.x"}, "g", 65, payloadKey, noop); err == nil {
+		t.Fatal("65 lanes accepted")
+	}
+	if err := bus.SubscribeKeyed(t.Context(), []string{"w.x"}, "g", 2, nil, noop); err == nil {
+		t.Fatal("several lanes without a key resolver accepted")
+	}
+}
+
+func TestNATSBus_KeyedLanesKeepSaturatedMessagesAlive(t *testing.T) {
+	grace := natsAckGrace
+	natsAckGrace = 200 * time.Millisecond
+	t.Cleanup(func() { natsAckGrace = grace })
+	t.Setenv("EVENTBUS_HANDLER_TIMEOUT", "300ms")
+
+	bus := newTestNATSBus(t)
+	topic := "w.saturated"
+	const messages = 64
+	var mu sync.Mutex
+	var order []string
+	seen := map[string][]int{}
+	allHandled := make(chan struct{})
+	subscribeInBackground(t, t.Context(), bus, topic, "saturated", 8, payloadKey, func(_ context.Context, msg Message) error {
+		p := string(msg.Payload)
+		mu.Lock()
+		order = append(order, p)
+		seen[p] = append(seen[p], msg.Attempt)
+		if len(seen) == messages && len(seen[p]) == 1 {
+			close(allHandled)
+		}
+		mu.Unlock()
+		time.Sleep(150 * time.Millisecond)
+		return nil
+	})
+	ackWait := waitForConsumer(t, bus, bus.durable("saturated", []string{topic}))
+	for i := 0; i < messages; i++ {
+		p := fmt.Sprintf("a%02d", i)
+		if err := bus.Publish(t.Context(), topic, "a", []byte(p)); err != nil {
+			t.Fatalf("publish %s: %v", p, err)
+		}
+	}
+	select {
+	case <-allHandled:
+	case <-time.After(20 * time.Second):
+		t.Fatal("not every saturated-lane message was handled")
+	}
+	// Give already-prefetched redeliveries time to reach the handler too.
+	time.Sleep(2 * ackWait)
+	mu.Lock()
+	defer mu.Unlock()
+	for i := 0; i < messages; i++ {
+		p := fmt.Sprintf("a%02d", i)
+		if got := seen[p]; len(got) != 1 || got[0] != 1 {
+			t.Errorf("%s handled with attempts %v, want exactly one first delivery", p, got)
+		}
+		if order[i] != p {
+			t.Errorf("message %d = %s, want %s", i, order[i], p)
+		}
+	}
+	if len(order) != messages {
+		t.Errorf("handled %d messages, want %d", len(order), messages)
+	}
+}