feat: enforce signature-scoped drainage authorization before SMS submission

gateway/internal/upstream/submit.go

@@ -32,7 +32,7 @@ func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.Su
 		// 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)
-	})
+	}, func() error { return m.authorizeDrainage(ctx, cmd) })
 	return result, err
 }
 
@@ -40,6 +40,7 @@ func (p *connectionPool) submit(
 	ctx context.Context,
 	cmd queue.SubmitCommand,
 	onSegment func(queue.SubmitSegmentResult) error,
+	authorize ...func() error,
 ) (final queue.SubmitResult, finalErr error) {
 	defer func() {
 		for _, segment := range final.Segments {
@@ -67,6 +68,23 @@ func (p *connectionPool) submit(
 			result.Segments = segments
 			return result, err
 		}
+		for _, check := range authorize {
+			if err := check(); err != nil {
+				release()
+				code := "DRNCHK"
+				status := "rejected"
+				if _, business := err.(*drainageBusinessRejection); business {
+					code = "DRN"
+				} else if len(segments) == 0 {
+					// No bytes were submitted. Let the existing durable worker retry
+					// a technical outage with its bounded backoff/dead-letter policy.
+					status = ""
+				}
+				result := submitResult(cmd, 0, "", status, code, err.Error())
+				result.Segments = segments
+				return result, err
+			}
+		}
 		supplierStartedAt := time.Now()
 		seq, gatewayMessageID, result, err := conn.submitPart(ctx, cmd, part)
 		metrics.ObserveSubmitStage("supplier_rtt", err == nil, time.Since(supplierStartedAt))