fix: use PublishAsync in pipeline to remove per-publish NATS ceiling

M11 soak showed ~470/s per instance vs the 5k/s target. Root cause:
js.Publish (sync) blocks for the JetStream PubAck (~2ms), serialising
every pipeline goroutine on a single NATS round-trip:

  ceiling = 1 / publish_latency = 1 / 0.002s ≈ 500/s

Switch the hot path to PublishAsync, which returns a PubAckFuture
immediately after JetStream accepts the message into its internal
queue. The circuit breaker still wraps the *submission* (not the
ack), so a stalled JetStream still surfaces to the breaker within
the configured timeout.

A new observeAsyncAck goroutine records publish latency on the
future's Ok channel and logs a warn on Err. This moves failure
detection from synchronous (in the hot path) to async (in a
goroutine), so Process returns accepted=true once JetStream has
buffered the message, not once it is persisted. At-most-once
semantics are unchanged: the broker still applies the per-subject
dedup and we use the JS acknowledgement to update circuit state.

Interface change: natsPublisher.PublishAsync now returns
(nats.PubAckFuture, error) instead of just error. Test fake in
cmd/ingestd/http_test.go updated to match.

cmd/ingestd/http_test.go

@@ -20,6 +20,7 @@ import (
 	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
 	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
 	pipeline "git3.techno-world.net/lrosales/broad-announce/internal/pipeline"
+	"github.com/nats-io/nats.go"
 )
 
 // fakePublisher records subjects+payloads.
@@ -32,16 +33,17 @@ type fakePub struct {
 	payload []byte
 }
 
-func (f *fakePublisher) PublishAsync(subj string, data []byte) error {
+func (f *fakePublisher) PublishAsync(subj string, data []byte) (nats.PubAckFuture, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.items = append(f.items, fakePub{subj, append([]byte(nil), data...)})
-	return nil
+	return nil, nil
 }
 
 // Publish is synchronous; same as PublishAsync for the test fake.
 func (f *fakePublisher) Publish(subj string, data []byte) error {
-	return f.PublishAsync(subj, data)
+	_, err := f.PublishAsync(subj, data)
+	return err
 }
 
 // stubLimiter always allows.

internal/pipeline/pipeline.go

@@ -246,11 +246,29 @@ func (d *Deps) Process(ctx context.Context, body []byte, sig string) Result {
 	start := time.Now()
 	var publishErr error
 	if d.CircuitBreaker != nil {
+		// M11 fix: PublishAsync returns a future immediately. The CB wraps
+		// the *submission* (not the ack) so a JetStream stall still surfaces
+		// to the circuit breaker within the configured timeout. The actual
+		// ack is observed in a fire-and-forget goroutine.
+		var fut nats.PubAckFuture
 		publishErr = d.CircuitBreaker.Do(ctx, func() error {
-			return d.JetStream.Publish(subject, payload)
+			f, err := d.JetStream.PublishAsync(subject, payload)
+			if err != nil {
+				return err
+			}
+			fut = f
+			return nil
 		})
+		if publishErr == nil && fut != nil {
+			go observeAsyncAck(fut, d, a.SourceID, subject, start)
+		}
 	} else {
-		publishErr = d.JetStream.Publish(subject, payload)
+		fut, err := d.JetStream.PublishAsync(subject, payload)
+		if err != nil {
+			publishErr = err
+		} else if fut != nil {
+			go observeAsyncAck(fut, d, a.SourceID, subject, start)
+		}
 	}
 	if publishErr != nil {
 		if errors.Is(publishErr, circuitbreaker.ErrCircuitOpen) {
@@ -308,7 +326,10 @@ func (d *Deps) Now() time.Time {
 // natsPublisher is the minimal NATS interface the pipeline needs.
 type natsPublisher interface {
 	Publish(subj string, data []byte) error
-	PublishAsync(subj string, data []byte) error
+	// PublishAsync submits to JetStream's internal queue and returns a
+	// future that resolves when the broker acks persistence. Callers
+	// observe the future asynchronously to avoid blocking the hot path.
+	PublishAsync(subj string, data []byte) (nats.PubAckFuture, error)
 }
 
 // jsPublisher adapts nats.JetStreamContext to natsPublisher.
@@ -319,9 +340,8 @@ func (j *jsPublisher) Publish(subj string, data []byte) error {
 	return err
 }
 
-func (j *jsPublisher) PublishAsync(subj string, data []byte) error {
-	_, err := j.js.PublishAsync(subj, data)
-	return err
+func (j *jsPublisher) PublishAsync(subj string, data []byte) (nats.PubAckFuture, error) {
+	return j.js.PublishAsync(subj, data)
 }
 
 // NewNatsPublisher constructs a natsPublisher from a JetStream context.
@@ -329,6 +349,25 @@ func NewNatsPublisher(js nats.JetStreamContext) natsPublisher {
 	return &jsPublisher{js: js}
 }
 
+// observeAsyncAck blocks on the PubAckFuture and records publish latency
+// or a warn-level log on failure. Runs in its own goroutine so the hot path
+// returns immediately. Latency is measured from start to broker ack.
+func observeAsyncAck(fut nats.PubAckFuture, d *Deps, sourceID, subject string, sentAt time.Time) {
+	if fut == nil {
+		return
+	}
+	select {
+	case <-fut.Ok():
+		if d != nil && d.Metrics != nil {
+			d.Metrics.PublishLatency.WithLabelValues(sourceID).Observe(time.Since(sentAt).Seconds())
+		}
+	case err := <-fut.Err():
+		if d != nil && d.Logger != nil {
+			d.Logger.Warn("async publish failed", "subject", subject, "source_id", sourceID, "err", err)
+		}
+	}
+}
+
 // verifyHMAC parses `X-BA-Signature: t=<unix>,v1=<hex>` and checks
 // HMAC-SHA256(secret, "<unix>.<body>") == hex. Replay window: 5 min.
 // Exported so HTTP handlers can call it directly; gRPC passes sig="".