perf(cmpp): decouple supplier result callbacks

gateway/internal/metrics/metrics.go

@@ -43,12 +43,18 @@ type Snapshot struct {
 	SubmitWorkerUp           bool
 	SubmitWorkerConcurrency  int
 	SubmitWorkerInFlight     int64
+	ResultWorkerUp           bool
+	ResultWorkerConcurrency  int
+	ResultWorkerInFlight     int64
 	InboundSubmitConcurrency int
 	InboundSubmitInFlight    int64
 	QueueAvailable           bool
 	QueuePending             int64
 	QueueLag                 int64
 	QueueOldestAgeSeconds    float64
+	ResultQueueAvailable     bool
+	ResultQueuePending       int64
+	ResultQueueLag           int64
 }
 
 type SnapshotFunc func(context.Context) Snapshot
@@ -118,12 +124,18 @@ func Handler(load SnapshotFunc) http.Handler {
 		fmt.Fprintf(response, "# HELP cmpp_gateway_downstream_connections Authenticated client connections.\n# TYPE cmpp_gateway_downstream_connections gauge\ncmpp_gateway_downstream_connections %d\n", snapshot.DownstreamConnected)
 		fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_up Whether the submit worker was initialized.\n# TYPE cmpp_gateway_submit_worker_up gauge\ncmpp_gateway_submit_worker_up %d\n", boolNumber(snapshot.SubmitWorkerUp))
 		fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_slots Configured and active bounded submit worker slots.\n# TYPE cmpp_gateway_submit_worker_slots gauge\ncmpp_gateway_submit_worker_slots{state=\"configured\"} %d\ncmpp_gateway_submit_worker_slots{state=\"in_flight\"} %d\n", snapshot.SubmitWorkerConcurrency, snapshot.SubmitWorkerInFlight)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_result_callback_worker_up Whether the asynchronous result callback worker was initialized.\n# TYPE cmpp_gateway_result_callback_worker_up gauge\ncmpp_gateway_result_callback_worker_up %d\n", boolNumber(snapshot.ResultWorkerUp))
+		fmt.Fprintf(response, "# HELP cmpp_gateway_result_callback_worker_slots Configured and active result callback worker slots.\n# TYPE cmpp_gateway_result_callback_worker_slots gauge\ncmpp_gateway_result_callback_worker_slots{state=\"configured\"} %d\ncmpp_gateway_result_callback_worker_slots{state=\"in_flight\"} %d\n", snapshot.ResultWorkerConcurrency, snapshot.ResultWorkerInFlight)
 		fmt.Fprintf(response, "# HELP cmpp_gateway_inbound_submit_slots Configured and active authenticated client Submit slots.\n# TYPE cmpp_gateway_inbound_submit_slots gauge\ncmpp_gateway_inbound_submit_slots{state=\"configured\"} %d\ncmpp_gateway_inbound_submit_slots{state=\"in_flight\"} %d\n", snapshot.InboundSubmitConcurrency, snapshot.InboundSubmitInFlight)
 		if snapshot.QueueAvailable {
 			fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_pending Pending entries owned by the consumer group.\n# TYPE cmpp_gateway_submit_queue_pending gauge\ncmpp_gateway_submit_queue_pending %d\n", snapshot.QueuePending)
 			fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_lag Undelivered entries for the consumer group.\n# TYPE cmpp_gateway_submit_queue_lag gauge\ncmpp_gateway_submit_queue_lag %d\n", snapshot.QueueLag)
 			fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_oldest_pending_age_seconds Age of the oldest pending entry.\n# TYPE cmpp_gateway_submit_queue_oldest_pending_age_seconds gauge\ncmpp_gateway_submit_queue_oldest_pending_age_seconds %f\n", snapshot.QueueOldestAgeSeconds)
 		}
+		if snapshot.ResultQueueAvailable {
+			fmt.Fprintf(response, "# HELP cmpp_gateway_result_outbox_pending Pending result callbacks owned by the consumer group.\n# TYPE cmpp_gateway_result_outbox_pending gauge\ncmpp_gateway_result_outbox_pending %d\n", snapshot.ResultQueuePending)
+			fmt.Fprintf(response, "# HELP cmpp_gateway_result_outbox_lag Undelivered result callbacks for the consumer group.\n# TYPE cmpp_gateway_result_outbox_lag gauge\ncmpp_gateway_result_outbox_lag %d\n", snapshot.ResultQueueLag)
+		}
 	})
 }
 

gateway/internal/metrics/metrics_test.go

@@ -16,7 +16,7 @@ func TestMetricsHandlerExportsOnlyAggregateGatewayState(t *testing.T) {
 	request := httptest.NewRequest(http.MethodGet, "/metrics", nil)
 	response := httptest.NewRecorder()
 	Handler(func(context.Context) Snapshot {
-		return Snapshot{UpstreamDesired: 2, UpstreamConnected: 1, DownstreamConnected: 3, SubmitWorkerUp: true, SubmitWorkerConcurrency: 64, SubmitWorkerInFlight: 7, InboundSubmitConcurrency: 96, InboundSubmitInFlight: 9, QueueAvailable: true, QueuePending: 4, QueueLag: 5, QueueOldestAgeSeconds: 12}
+		return Snapshot{UpstreamDesired: 2, UpstreamConnected: 1, DownstreamConnected: 3, SubmitWorkerUp: true, SubmitWorkerConcurrency: 64, SubmitWorkerInFlight: 7, ResultWorkerUp: true, ResultWorkerConcurrency: 32, ResultWorkerInFlight: 5, InboundSubmitConcurrency: 96, InboundSubmitInFlight: 9, QueueAvailable: true, QueuePending: 4, QueueLag: 5, QueueOldestAgeSeconds: 12, ResultQueueAvailable: true, ResultQueuePending: 6, ResultQueueLag: 8}
 	}).ServeHTTP(response, request)
 
 	body := response.Body.String()
@@ -28,6 +28,10 @@ func TestMetricsHandlerExportsOnlyAggregateGatewayState(t *testing.T) {
 		`cmpp_gateway_submit_stage_duration_seconds_count{stage="rate_limit_wait",result="success"} 1`,
 		`cmpp_gateway_submit_worker_slots{state="configured"} 64`,
 		`cmpp_gateway_submit_worker_slots{state="in_flight"} 7`,
+		`cmpp_gateway_result_callback_worker_slots{state="configured"} 32`,
+		`cmpp_gateway_result_callback_worker_slots{state="in_flight"} 5`,
+		`cmpp_gateway_result_outbox_pending 6`,
+		`cmpp_gateway_result_outbox_lag 8`,
 		`cmpp_gateway_inbound_submit_slots{state="configured"} 96`,
 		`cmpp_gateway_inbound_submit_slots{state="in_flight"} 9`,
 	} {

gateway/internal/resultoutbox/outbox.go

@@ -0,0 +1,260 @@
+package resultoutbox
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"cmpp-platform/gateway/internal/queue"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	defaultStream    = "gateway.submit.results"
+	defaultGroup     = "cmpp-api-callback"
+	defaultConsumer  = "gateway-1"
+	defaultDedupeTTL = 7 * 24 * time.Hour
+)
+
+var publishScript = redis.NewScript(`
+local inserted = redis.call('SET', KEYS[2], '1', 'NX', 'EX', ARGV[1])
+if inserted then
+  redis.call('XADD', KEYS[1], '*', 'data', ARGV[2])
+  return 1
+end
+return 0
+`)
+
+var publishAndAckScript = redis.NewScript(`
+local inserted = redis.call('SET', KEYS[2], '1', 'NX', 'EX', ARGV[1])
+if inserted then
+  redis.call('XADD', KEYS[1], '*', 'data', ARGV[2])
+end
+redis.call('XACK', KEYS[3], ARGV[3], ARGV[4])
+if inserted then return 1 end
+return 0
+`)
+
+type Event struct {
+	SchemaVersion string          `json:"schemaVersion"`
+	EventID       string          `json:"eventId"`
+	EventType     string          `json:"eventType"`
+	Path          string          `json:"path"`
+	TraceID       string          `json:"traceId,omitempty"`
+	MessageID     string          `json:"messageId"`
+	ChannelID     string          `json:"channelId"`
+	SubmitID      string          `json:"submitId"`
+	Payload       json.RawMessage `json:"payload"`
+	CreatedAt     time.Time       `json:"createdAt"`
+}
+
+type Outbox struct {
+	Redis       *redis.Client
+	Stream      string
+	Group       string
+	Consumer    string
+	DedupeTTL   time.Duration
+	APIBaseURL  string
+	HTTPTimeout time.Duration
+	Concurrency int
+	MinIdle     time.Duration
+	inFlight    atomic.Int64
+}
+
+func New(client *redis.Client) *Outbox {
+	return &Outbox{Redis: client}
+}
+
+func (o *Outbox) PublishSubmitSegment(ctx context.Context, command queue.SubmitCommand, segment queue.SubmitSegmentResult) error {
+	payload := struct {
+		queue.Envelope
+		SubmitID string `json:"submitId,omitempty"`
+		queue.SubmitSegmentResult
+	}{
+		Envelope:            command.Envelope,
+		SubmitID:            command.SubmitID,
+		SubmitSegmentResult: segment,
+	}
+	event, err := newEvent(
+		fmt.Sprintf("submit:%s:segment:%d", command.SubmitID, segment.SegmentIndex),
+		"submit_segment_result",
+		"/gateway/events/submit-segment-result",
+		command,
+		payload,
+	)
+	if err != nil {
+		return err
+	}
+	return o.publish(ctx, event)
+}
+
+func (o *Outbox) PublishSubmitResultAndAck(
+	ctx context.Context,
+	commandStream string,
+	commandGroup string,
+	commandMessageID string,
+	command queue.SubmitCommand,
+	result queue.SubmitResult,
+) error {
+	if o.Redis == nil {
+		return fmt.Errorf("result Outbox Redis client is required")
+	}
+	event, err := newEvent(
+		fmt.Sprintf("submit:%s:aggregate", command.SubmitID),
+		"submit_result",
+		"/gateway/events/submit-result",
+		command,
+		result,
+	)
+	if err != nil {
+		return err
+	}
+	data, err := json.Marshal(event)
+	if err != nil {
+		return err
+	}
+	// XADD and command XACK share one Redis script so a process crash cannot leave
+	// an acknowledged supplier command without its aggregate result in the Outbox.
+	_, err = publishAndAckScript.Run(
+		ctx,
+		o.Redis,
+		[]string{o.stream(), o.dedupeKey(event.EventID), commandStream},
+		int64(o.dedupeTTL().Seconds()),
+		string(data),
+		commandGroup,
+		commandMessageID,
+	).Result()
+	return err
+}
+
+func (o *Outbox) PublishSubmitResult(ctx context.Context, command queue.SubmitCommand, result queue.SubmitResult) error {
+	event, err := newEvent(
+		fmt.Sprintf("submit:%s:aggregate", command.SubmitID),
+		"submit_result",
+		"/gateway/events/submit-result",
+		command,
+		result,
+	)
+	if err != nil {
+		return err
+	}
+	return o.publish(ctx, event)
+}
+
+func (o *Outbox) publish(ctx context.Context, event Event) error {
+	if o.Redis == nil {
+		return fmt.Errorf("result Outbox Redis client is required")
+	}
+	data, err := json.Marshal(event)
+	if err != nil {
+		return err
+	}
+	_, err = publishScript.Run(
+		ctx,
+		o.Redis,
+		[]string{o.stream(), o.dedupeKey(event.EventID)},
+		int64(o.dedupeTTL().Seconds()),
+		string(data),
+	).Result()
+	return err
+}
+
+func newEvent(eventID string, eventType string, path string, command queue.SubmitCommand, payload any) (Event, error) {
+	data, err := json.Marshal(payload)
+	if err != nil {
+		return Event{}, err
+	}
+	var object map[string]interface{}
+	if err := json.Unmarshal(data, &object); err != nil {
+		return Event{}, err
+	}
+	// The API stores the deterministic ID on the submit attempt after successful
+	// processing, making callback redelivery idempotent across Gateway restarts.
+	object["eventId"] = eventID
+	data, err = json.Marshal(object)
+	if err != nil {
+		return Event{}, err
+	}
+	return Event{
+		SchemaVersion: queue.SchemaVersion,
+		EventID:       eventID,
+		EventType:     eventType,
+		Path:          path,
+		TraceID:       command.TraceID,
+		MessageID:     command.MessageID,
+		ChannelID:     command.ChannelID,
+		SubmitID:      command.SubmitID,
+		Payload:       data,
+		CreatedAt:     time.Now().UTC(),
+	}, nil
+}
+
+func EventFromStreamValues(values map[string]interface{}) (Event, error) {
+	raw, ok := values["data"]
+	if !ok {
+		return Event{}, fmt.Errorf("result Outbox data field is required")
+	}
+	var data string
+	switch value := raw.(type) {
+	case string:
+		data = value
+	case []byte:
+		data = string(value)
+	default:
+		data = fmt.Sprint(value)
+	}
+	var event Event
+	if err := json.Unmarshal([]byte(data), &event); err != nil {
+		return Event{}, err
+	}
+	if event.SchemaVersion != queue.SchemaVersion || event.EventID == "" || event.MessageID == "" || event.SubmitID == "" {
+		return Event{}, fmt.Errorf("invalid result Outbox envelope")
+	}
+	if event.Path != "/gateway/events/submit-result" && event.Path != "/gateway/events/submit-segment-result" {
+		return Event{}, fmt.Errorf("unsupported result Outbox path %q", event.Path)
+	}
+	if len(event.Payload) == 0 {
+		return Event{}, fmt.Errorf("result Outbox payload is required")
+	}
+	return event, nil
+}
+
+func (o *Outbox) stream() string {
+	if strings.TrimSpace(o.Stream) != "" {
+		return o.Stream
+	}
+	return defaultStream
+}
+
+func (o *Outbox) group() string {
+	if strings.TrimSpace(o.Group) != "" {
+		return o.Group
+	}
+	return defaultGroup
+}
+
+func (o *Outbox) consumer() string {
+	if strings.TrimSpace(o.Consumer) != "" {
+		return o.Consumer
+	}
+	return defaultConsumer
+}
+
+func (o *Outbox) dedupeTTL() time.Duration {
+	if o.DedupeTTL > 0 {
+		return o.DedupeTTL
+	}
+	return defaultDedupeTTL
+}
+
+func (o *Outbox) dedupeKey(eventID string) string {
+	return o.stream() + ":dedupe:" + eventID
+}
+
+func (o *Outbox) StreamName() string { return o.stream() }
+func (o *Outbox) GroupName() string  { return o.group() }
+func (o *Outbox) InFlight() int64    { return o.inFlight.Load() }

gateway/internal/resultoutbox/outbox_test.go

@@ -0,0 +1,129 @@
+package resultoutbox
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"cmpp-platform/gateway/internal/queue"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+func TestPublishSubmitSegmentIsIdempotent(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	outbox := New(client)
+	command := testCommand()
+	segment := queue.SubmitSegmentResult{SegmentTotal: 1, SegmentIndex: 1, SequenceID: 7, GatewayMessageID: "88", SubmitStatus: "accepted"}
+
+	if err := outbox.PublishSubmitSegment(context.Background(), command, segment); err != nil {
+		t.Fatalf("first publish: %v", err)
+	}
+	if err := outbox.PublishSubmitSegment(context.Background(), command, segment); err != nil {
+		t.Fatalf("duplicate publish: %v", err)
+	}
+	if got := client.XLen(context.Background(), outbox.StreamName()).Val(); got != 1 {
+		t.Fatalf("stream length = %d, want 1", got)
+	}
+}
+
+func TestPublishAggregateAndCommandAckAreAtomicAndIdempotent(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	ctx := context.Background()
+	commandStream := "gateway.submit.commands"
+	commandGroup := "cmpp-gateway"
+	if err := client.XGroupCreateMkStream(ctx, commandStream, commandGroup, "0").Err(); err != nil {
+		t.Fatalf("create command group: %v", err)
+	}
+	commandID, err := client.XAdd(ctx, &redis.XAddArgs{Stream: commandStream, Values: map[string]interface{}{"data": "command"}}).Result()
+	if err != nil {
+		t.Fatalf("add command: %v", err)
+	}
+	if _, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: commandGroup, Consumer: "gateway-1", Streams: []string{commandStream, ">"}, Count: 1}).Result(); err != nil {
+		t.Fatalf("claim command: %v", err)
+	}
+	outbox := New(client)
+	command := testCommand()
+	result := queue.SubmitResult{Envelope: command.Envelope, SubmitID: command.SubmitID, GatewayMessageID: "99", SubmitStatus: "accepted"}
+
+	for attempt := 0; attempt < 2; attempt++ {
+		if err := outbox.PublishSubmitResultAndAck(ctx, commandStream, commandGroup, commandID, command, result); err != nil {
+			t.Fatalf("publish attempt %d: %v", attempt+1, err)
+		}
+	}
+	pending, err := client.XPending(ctx, commandStream, commandGroup).Result()
+	if err != nil {
+		t.Fatalf("command pending: %v", err)
+	}
+	if pending.Count != 0 {
+		t.Fatalf("command pending = %d, want 0", pending.Count)
+	}
+	if got := client.XLen(ctx, outbox.StreamName()).Val(); got != 1 {
+		t.Fatalf("result stream length = %d, want 1", got)
+	}
+}
+
+func TestCallbackWorkerRetriesAndOnlyDeletesAfterSuccess(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	var calls atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		if request.Header.Get("X-CMPP-Result-Event-ID") == "" {
+			t.Error("missing result event id header")
+		}
+		if calls.Add(1) == 1 {
+			http.Error(response, "temporary failure", http.StatusServiceUnavailable)
+			return
+		}
+		response.WriteHeader(http.StatusCreated)
+	}))
+	defer server.Close()
+
+	outbox := New(client)
+	outbox.APIBaseURL = server.URL
+	outbox.MinIdle = 10 * time.Millisecond
+	outbox.Concurrency = 1
+	command := testCommand()
+	if err := outbox.PublishSubmitSegment(context.Background(), command, queue.SubmitSegmentResult{
+		SegmentTotal: 1, SegmentIndex: 1, SequenceID: 7, GatewayMessageID: "88", SubmitStatus: "accepted",
+	}); err != nil {
+		t.Fatalf("publish segment: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- outbox.Run(ctx) }()
+	deadline := time.Now().Add(5 * time.Second)
+	for calls.Load() < 2 || client.XLen(context.Background(), outbox.StreamName()).Val() != 0 {
+		if time.Now().After(deadline) {
+			t.Fatalf("calls=%d streamLength=%d", calls.Load(), client.XLen(context.Background(), outbox.StreamName()).Val())
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("worker did not stop")
+	}
+}
+
+func testCommand() queue.SubmitCommand {
+	return queue.SubmitCommand{
+		Envelope: queue.Envelope{
+			SchemaVersion: queue.SchemaVersion,
+			MessageType:   queue.MessageTypeSubmitCommand,
+			TraceID:       "trace-1",
+			MessageID:     "message-1",
+			ChannelID:     "channel-1",
+			CreatedAt:     time.Now().UTC(),
+		},
+		SubmitID: "submit-1",
+	}
+}

gateway/internal/resultoutbox/worker.go

@@ -0,0 +1,275 @@
+package resultoutbox
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"cmpp-platform/gateway/internal/metrics"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	// Eight callbacks matched the test VM's API/PostgreSQL capacity through
+	// 40 rps. A larger default caused callback bursts to contend with inbound
+	// persistence; operators can still raise it after measuring both queues.
+	defaultConcurrency = 8
+	// Keep the reclaim threshold above the callback timeout. Otherwise another
+	// Gateway replica could reclaim a still-running callback and execute the same
+	// business transition concurrently before the API stores its idempotency ID.
+	defaultMinIdle     = 30 * time.Second
+	defaultBlock       = 2 * time.Second
+	defaultHTTPTimeout = 10 * time.Second
+)
+
+var acknowledgeAndDeleteScript = redis.NewScript(`
+redis.call('XACK', KEYS[1], ARGV[1], ARGV[2])
+redis.call('XDEL', KEYS[1], ARGV[2])
+return 1
+`)
+
+func (o *Outbox) Run(ctx context.Context) error {
+	if o.Redis == nil {
+		return fmt.Errorf("result Outbox Redis client is required")
+	}
+	if strings.TrimSpace(o.APIBaseURL) == "" {
+		return fmt.Errorf("result Outbox API base URL is required")
+	}
+	if err := o.ensureGroup(ctx); err != nil {
+		return err
+	}
+	pool := newCallbackPool(ctx, o, o.concurrency())
+	defer pool.wait()
+	for {
+		if err := o.recoverPending(ctx, pool); err != nil && ctx.Err() == nil {
+			log.Printf("gateway result Outbox pending recovery failed: %v", err)
+			sleep(ctx, time.Second)
+			continue
+		}
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		if err := o.consumeOnce(ctx, pool); err != nil && ctx.Err() == nil {
+			log.Printf("gateway result Outbox consume failed: %v", err)
+			sleep(ctx, time.Second)
+			continue
+		}
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+	}
+}
+
+func (o *Outbox) ensureGroup(ctx context.Context) error {
+	err := o.Redis.XGroupCreateMkStream(ctx, o.stream(), o.group(), "0").Err()
+	if err == nil || strings.Contains(err.Error(), "BUSYGROUP") {
+		return nil
+	}
+	return err
+}
+
+func (o *Outbox) consumeOnce(ctx context.Context, pool *callbackPool) error {
+	available, err := pool.waitForCapacity(ctx)
+	if err != nil {
+		return err
+	}
+	streams, err := o.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{
+		Group: o.group(), Consumer: o.consumer(), Streams: []string{o.stream(), ">"},
+		Count: int64(available), Block: defaultBlock,
+	}).Result()
+	if errors.Is(err, redis.Nil) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	for _, stream := range streams {
+		for _, message := range stream.Messages {
+			if !pool.dispatch(message) {
+				return fmt.Errorf("gateway result Outbox capacity accounting mismatch")
+			}
+		}
+	}
+	return nil
+}
+
+func (o *Outbox) recoverPending(ctx context.Context, pool *callbackPool) error {
+	if pool.available() == 0 {
+		return nil
+	}
+	messages, _, err := o.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{
+		Stream: o.stream(), Group: o.group(), Consumer: o.consumer(), MinIdle: o.minIdle(),
+		Start: "0-0", Count: int64(pool.available()),
+	}).Result()
+	if errors.Is(err, redis.Nil) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	for _, message := range messages {
+		if !pool.dispatch(message) {
+			return fmt.Errorf("gateway result Outbox recovery capacity accounting mismatch")
+		}
+	}
+	return nil
+}
+
+func (o *Outbox) processMessage(ctx context.Context, message redis.XMessage) error {
+	event, err := EventFromStreamValues(message.Values)
+	if err != nil {
+		// Malformed internal events cannot be delivered. Keep them pending for operator
+		// evidence instead of ACKing and silently losing a supplier result.
+		return err
+	}
+	startedAt := time.Now()
+	err = o.post(ctx, event)
+	metrics.ObserveSubmitStage("api_callback", err == nil, time.Since(startedAt))
+	if err != nil {
+		return err
+	}
+	// Result events have a bounded dedupe key, so successful callbacks can be
+	// ACKed and deleted atomically instead of turning the Outbox into an archive.
+	return acknowledgeAndDeleteScript.Run(
+		ctx,
+		o.Redis,
+		[]string{o.stream()},
+		o.group(),
+		message.ID,
+	).Err()
+}
+
+func (o *Outbox) post(ctx context.Context, event Event) error {
+	client := &http.Client{Timeout: o.httpTimeout()}
+	req, err := http.NewRequestWithContext(
+		ctx,
+		http.MethodPost,
+		strings.TrimRight(o.APIBaseURL, "/")+event.Path,
+		bytes.NewReader(event.Payload),
+	)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-CMPP-Result-Event-ID", event.EventID)
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+		return fmt.Errorf("result callback returned %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
+	}
+	return nil
+}
+
+type callbackPool struct {
+	ctx       context.Context
+	outbox    *Outbox
+	slots     chan struct{}
+	completed chan struct{}
+	group     sync.WaitGroup
+	mu        sync.Mutex
+	active    map[string]struct{}
+}
+
+func newCallbackPool(ctx context.Context, outbox *Outbox, concurrency int) *callbackPool {
+	return &callbackPool{
+		ctx: ctx, outbox: outbox, slots: make(chan struct{}, concurrency), completed: make(chan struct{}, concurrency), active: make(map[string]struct{}),
+	}
+}
+
+func (p *callbackPool) available() int { return cap(p.slots) - len(p.slots) }
+
+func (p *callbackPool) waitForCapacity(ctx context.Context) (int, error) {
+	for p.available() == 0 {
+		select {
+		case <-ctx.Done():
+			return 0, ctx.Err()
+		case <-p.completed:
+		}
+	}
+	return p.available(), nil
+}
+
+func (p *callbackPool) dispatch(message redis.XMessage) bool {
+	p.mu.Lock()
+	if _, exists := p.active[message.ID]; exists {
+		p.mu.Unlock()
+		return true
+	}
+	select {
+	case p.slots <- struct{}{}:
+		p.active[message.ID] = struct{}{}
+		p.outbox.inFlight.Add(1)
+		p.group.Add(1)
+		p.mu.Unlock()
+	case <-p.ctx.Done():
+		p.mu.Unlock()
+		return false
+	default:
+		p.mu.Unlock()
+		return false
+	}
+	go func() {
+		defer func() {
+			p.mu.Lock()
+			delete(p.active, message.ID)
+			p.mu.Unlock()
+			<-p.slots
+			p.outbox.inFlight.Add(-1)
+			select {
+			case p.completed <- struct{}{}:
+			default:
+			}
+			p.group.Done()
+		}()
+		if err := p.outbox.processMessage(p.ctx, message); err != nil {
+			log.Printf("gateway result Outbox event %s failed: %v", message.ID, err)
+		}
+	}()
+	return true
+}
+
+func (p *callbackPool) wait() { p.group.Wait() }
+
+func (o *Outbox) concurrency() int {
+	if o.Concurrency > 0 {
+		return min(o.Concurrency, 1024)
+	}
+	return defaultConcurrency
+}
+
+func (o *Outbox) minIdle() time.Duration {
+	if o.MinIdle > 0 {
+		return o.MinIdle
+	}
+	return defaultMinIdle
+}
+
+func (o *Outbox) httpTimeout() time.Duration {
+	if o.HTTPTimeout > 0 {
+		return o.HTTPTimeout
+	}
+	return defaultHTTPTimeout
+}
+
+func (o *Outbox) ConfiguredConcurrency() int { return o.concurrency() }
+
+func sleep(ctx context.Context, duration time.Duration) {
+	timer := time.NewTimer(duration)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+	case <-timer.C:
+	}
+}

gateway/internal/submitworker/worker.go

@@ -36,6 +36,7 @@ type Worker struct {
 	Limiter          ratelimit.Limiter
 	Submit           func(context.Context, queue.SubmitCommand) (queue.SubmitResult, error)
 	ReportDeadLetter func(context.Context, DeadLetterEvent) error
+	ResultOutbox     SubmitResultOutbox
 	Stream           string
 	Group            string
 	Consumer         string
@@ -50,6 +51,17 @@ type Worker struct {
 	inFlight         atomic.Int64
 }
 
+type SubmitResultOutbox interface {
+	PublishSubmitResultAndAck(
+		context.Context,
+		string,
+		string,
+		string,
+		queue.SubmitCommand,
+		queue.SubmitResult,
+	) error
+}
+
 type DeadLetterEvent struct {
 	StreamMessageID string                 `json:"streamMessageId"`
 	TraceID         string                 `json:"traceId,omitempty"`
@@ -82,6 +94,9 @@ func (w *Worker) Run(ctx context.Context) error {
 	if w.Upstream == nil {
 		return fmt.Errorf("upstream manager is required")
 	}
+	if w.ResultOutbox == nil {
+		return fmt.Errorf("submit result Outbox is required")
+	}
 	pool := newMessageWorkPool(ctx, w, w.concurrency())
 	defer pool.wait()
 	for {
@@ -292,7 +307,8 @@ func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) err
 	if !command.CreatedAt.IsZero() {
 		metrics.ObserveSubmitStage("stream_wait", true, time.Since(command.CreatedAt))
 	}
-	if err := w.handleCommand(ctx, command); err != nil {
+	result, err := w.executeCommand(ctx, command)
+	if err != nil && (result.SubmitStatus == "" || result.SubmitStatus == "accepted") {
 		if ctx.Err() != nil {
 			return ctx.Err()
 		}
@@ -308,29 +324,50 @@ func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) err
 		}
 		return err
 	}
-	return w.ackAndClearFailure(ctx, message.ID)
+	if err := w.ResultOutbox.PublishSubmitResultAndAck(
+		ctx,
+		w.stream(),
+		w.group(),
+		message.ID,
+		command,
+		result,
+	); err != nil {
+		return fmt.Errorf("persist aggregate submit result: %w", err)
+	}
+	// The command was ACKed atomically with the aggregate result event. Clearing the
+	// auxiliary retry counter may lag without affecting delivery correctness.
+	w.clearFailure(ctx, message.ID)
+	return nil
 }
 
-func (w *Worker) handleCommand(ctx context.Context, command queue.SubmitCommand) error {
+func (w *Worker) executeCommand(ctx context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
 	startedAt := time.Now()
 	limitStartedAt := time.Now()
 	if w.Limiter != nil {
 		if _, err := w.Limiter.Wait(ctx, command.ChannelID, command.Route.RateLimitPerSecond); err != nil {
 			metrics.ObserveSubmitStage("rate_limit_wait", false, time.Since(limitStartedAt))
-			return err
+			return queue.SubmitResult{}, err
 		}
 	}
 	metrics.ObserveSubmitStage("rate_limit_wait", true, time.Since(limitStartedAt))
 	submit := w.Submit
 	if submit == nil {
 		if w.Upstream == nil {
-			return fmt.Errorf("upstream manager is required")
+			return queue.SubmitResult{}, fmt.Errorf("upstream manager is required")
 		}
 		submit = w.Upstream.Submit
 	}
 	result, err := submit(ctx, command)
 	accepted := err == nil && (result.SubmitStatus == "" || result.SubmitStatus == "accepted")
 	metrics.ObserveSubmit(accepted, time.Since(startedAt))
+	return result, err
+}
+
+// handleCommand remains a narrow compatibility seam for focused tests and the
+// control path. Stream consumption uses executeCommand so it can persist the
+// exact terminal result before acknowledging the command.
+func (w *Worker) handleCommand(ctx context.Context, command queue.SubmitCommand) error {
+	result, err := w.executeCommand(ctx, command)
 	if err != nil && (result.SubmitStatus == "" || result.SubmitStatus == "accepted") {
 		return err
 	}
@@ -456,10 +493,14 @@ func (w *Worker) ackAndClearFailure(ctx context.Context, messageID string) error
 	if err := w.Redis.XAck(ctx, w.stream(), w.group(), messageID).Err(); err != nil {
 		return err
 	}
+	w.clearFailure(ctx, messageID)
+	return nil
+}
+
+func (w *Worker) clearFailure(ctx context.Context, messageID string) {
 	if err := w.Redis.HDel(ctx, w.failureAttemptsKey(), messageID).Err(); err != nil {
 		w.logf("gateway submit worker clear failure %s failed: %v", messageID, err)
 	}
-	return nil
 }
 
 func (w *Worker) stream() string {

gateway/internal/submitworker/worker_test.go

@@ -17,6 +17,24 @@ type recordingLimiter struct {
 	called    bool
 }
 
+type acknowledgingResultOutbox struct {
+	client *redis.Client
+}
+
+func (o acknowledgingResultOutbox) PublishSubmitResultAndAck(
+	ctx context.Context,
+	commandStream string,
+	commandGroup string,
+	commandMessageID string,
+	_ queue.SubmitCommand,
+	_ queue.SubmitResult,
+) error {
+	if o.client == nil {
+		return nil
+	}
+	return o.client.XAck(ctx, commandStream, commandGroup, commandMessageID).Err()
+}
+
 func (l *recordingLimiter) Wait(_ context.Context, channelID string, rate int) (time.Duration, error) {
 	l.called = true
 	l.channelID = channelID
@@ -139,7 +157,8 @@ func TestMessageWorkPoolContinuouslyRefillsWithoutWaitingForSlowSibling(t *testi
 	startedC := make(chan struct{})
 	releaseA := make(chan struct{})
 	worker := &Worker{
-		Redis: client,
+		Redis:        client,
+		ResultOutbox: acknowledgingResultOutbox{client: client},
 		Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
 			switch command.ChannelID {
 			case "channel-a":
@@ -188,7 +207,7 @@ func TestMessageWorkPoolContinuouslyRefillsWithoutWaitingForSlowSibling(t *testi
 func TestMessageWorkPoolAcknowledgesFastMessageBeforeSlowSiblingCompletes(t *testing.T) {
 	mr := miniredis.RunT(t)
 	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
-	worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway"}
+	worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", ResultOutbox: acknowledgingResultOutbox{client: client}}
 	ctx := context.Background()
 	if err := worker.ensureGroup(ctx); err != nil {
 		t.Fatalf("ensureGroup: %v", err)
@@ -246,7 +265,7 @@ func TestMessageWorkPoolAcknowledgesFastMessageBeforeSlowSiblingCompletes(t *tes
 func TestPendingRecoveryDoesNotDuplicateAnActiveOrAlreadyAcknowledgedMessage(t *testing.T) {
 	mr := miniredis.RunT(t)
 	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
-	worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", MinIdle: time.Millisecond}
+	worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", MinIdle: time.Millisecond, ResultOutbox: acknowledgingResultOutbox{client: client}}
 	ctx := context.Background()
 	if err := worker.ensureGroup(ctx); err != nil {
 		t.Fatalf("ensureGroup: %v", err)

gateway/internal/upstream/manager.go

@@ -22,13 +22,21 @@ const (
 )
 
 type Manager struct {
-	APIBaseURL string
-	HTTPClient *http.Client
+	APIBaseURL             string
+	HTTPClient             *http.Client
+	SubmitSegmentPublisher SubmitSegmentPublisher
 
 	mu    sync.Mutex
 	conns map[string]*connectionPool
 }
 
+// SubmitSegmentPublisher persists each supplier response before the next long-message
+// segment is sent. The boundary is intentionally storage-only: HTTP callbacks belong to
+// the result Outbox worker and must not consume a supplier Submit window slot.
+type SubmitSegmentPublisher interface {
+	PublishSubmitSegment(context.Context, queue.SubmitCommand, queue.SubmitSegmentResult) error
+}
+
 type ConnectionState struct {
 	ChannelID              string `json:"channelId"`
 	ConnectionID           string `json:"connectionId"`

gateway/internal/upstream/submit.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	cmpp "github.com/bigwhite/gocmpp"
 	cmpputils "github.com/bigwhite/gocmpp/utils"
-	"log"
 	"strings"
 	"time"
 )
@@ -17,58 +16,30 @@ import (
 
 func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) {
 	if err := validateSubmitCommand(cmd); err != nil {
-		result := submitResult(cmd, 0, "", "rejected", "INVALID_COMMAND", err.Error())
-		if postErr := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); postErr != nil {
-			return result, postErr
-		}
-		return result, err
+		return submitResult(cmd, 0, "", "rejected", "INVALID_COMMAND", err.Error()), err
 	}
 
 	pool, err := m.connectionFor(cmd)
 	if err != nil {
-		result := submitResult(cmd, 0, "", "rejected", "CONNECT_FAILED", err.Error())
-		if postErr := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); postErr != nil {
-			return result, postErr
-		}
-		return result, err
+		return submitResult(cmd, 0, "", "rejected", "CONNECT_FAILED", err.Error()), err
 	}
 
-	result, err := pool.submit(ctx, cmd, func(segment queue.SubmitSegmentResult) {
-		payload := struct {
-			queue.Envelope
-			SubmitID string `json:"submitId,omitempty"`
-			queue.SubmitSegmentResult
-		}{
-			Envelope:            cmd.Envelope,
-			SubmitID:            cmd.SubmitID,
-			SubmitSegmentResult: segment,
-		}
-		callbackCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-		postErr := m.postSubmitCallback(callbackCtx, "/gateway/events/submit-segment-result", payload)
-		cancel()
-		if postErr != nil {
-			log.Printf(
-				"protocol_event protocol=cmpp direction=gateway_to_api event=submit_segment_result status=forward_failed channel_id=%s message_id=%s segment=%d/%d error=%q",
-				cmd.ChannelID, cmd.MessageID, segment.SegmentIndex, segment.SegmentTotal, postErr,
-			)
+	result, err := pool.submit(ctx, cmd, func(segment queue.SubmitSegmentResult) error {
+		if m.SubmitSegmentPublisher == nil {
+			return fmt.Errorf("submit segment result publisher is required")
 		}
+		// A segment result must reach durable local storage before the next segment.
+		// This cannot make the supplier/Redis boundary globally atomic, but it avoids
+		// holding the supplier slot for an API round trip and minimizes untracked sends.
+		return m.SubmitSegmentPublisher.PublishSubmitSegment(ctx, cmd, segment)
 	})
-	if err != nil {
-		if postErr := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); postErr != nil {
-			return result, postErr
-		}
-		return result, err
-	}
-	if err := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); err != nil {
-		return result, err
-	}
-	return result, nil
+	return result, err
 }
 
 func (p *connectionPool) submit(
 	ctx context.Context,
 	cmd queue.SubmitCommand,
-	onSegment func(queue.SubmitSegmentResult),
+	onSegment func(queue.SubmitSegmentResult) error,
 ) (queue.SubmitResult, error) {
 	parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content)
 	if err != nil {
@@ -95,7 +66,10 @@ func (p *connectionPool) submit(
 		segment := submitSegmentResult(part, seq, gatewayMessageID, result)
 		segments = append(segments, segment)
 		if onSegment != nil {
-			onSegment(segment)
+			if publishErr := onSegment(segment); publishErr != nil {
+				result.Segments = segments
+				return result, publishErr
+			}
 		}
 		if firstSequence == 0 {
 			firstSequence = seq
@@ -118,13 +92,6 @@ func (p *connectionPool) submit(
 	return result, nil
 }
 
-func (m *Manager) postSubmitCallback(ctx context.Context, path string, payload any) error {
-	startedAt := time.Now()
-	err := m.post(ctx, path, payload)
-	metrics.ObserveSubmitStage("api_callback", err == nil, time.Since(startedAt))
-	return err
-}
-
 func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, part submitPart) (uint32, string, queue.SubmitResult, error) {
 	rspCh := make(chan submitPartResponse, 1)
 	pkt := c.submitRequestPacket(cmd, part)