Fix long-running memory retention and stalled delivery

docs/adr/0001-route-delivery-boundaries.md

@@ -18,7 +18,13 @@ library auto-reconnect and gives each explicit broker connection a fresh client
 generation. A loss atomically invalidates that generation and every outstanding
 Push. With automatic reconnect disabled, a successful QoS 1 publish token is
 possible only after an observed PUBACK; any publish still outstanding at loss
-is retried on a newly constructed client. HTTP POST retries use a deterministic
+is retried on a newly constructed client. A broker that keeps its connection
+alive but withholds PUBACK also loses its generation after 30 seconds; the
+connector retains the unconfirmed message for retry. CONNECT and SUBSCRIBE
+waits are limited to 15 seconds. Beacon owns transport cleanup as well as Paho
+cleanup so a stalled handshake cannot retain an orphan socket.
+
+HTTP POST retries use a deterministic
 batch idempotency key, but remain at-least-once because the receiver decides
 whether to honor it. Receiver `Retry-After` guidance can lengthen, but never
 shorten, the connector's retry delay and is capped at one minute. Per-attempt
@@ -31,6 +37,13 @@ the upstream wire timestamp. Retention may prune pending delivery, so these
 guards are safety budgets rather than outage guarantees and must be sized from
 observed Envelope rate, size, and the supported disconnected interval.
 
+Connector shutdown unsubscribes before taking a fixed snapshot of buffered
+input. It flushes batches of at most 64 within one five-second grace period,
+including any batch interrupted by cancellation. Unpersisted messages at the
+deadline contribute to `intake_loss`; an active source cannot extend the drain.
+Count retention locates the oldest excess rows instead of scanning across the
+retained history on every saturated append.
+
 Remote-source, physical NMEA-endpoint, and confirmed-delivery retries run
 continuously with equal-jitter exponential delay from a 250 ms initial ceiling
 to one minute. Durable queue writes use 100 ms to one minute. This bounds

go.mod

@@ -19,6 +19,7 @@ require (
 	go.opentelemetry.io/otel/exporters/prometheus v0.66.0
 	go.opentelemetry.io/otel/metric v1.44.0
 	go.opentelemetry.io/otel/sdk/metric v1.44.0
+	golang.org/x/net v0.53.0
 	modernc.org/sqlite v1.53.0
 )
 
@@ -58,7 +59,6 @@ require (
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/exp v0.0.0-20240823005443-9b4947da3948 // indirect
-	golang.org/x/net v0.53.0 // indirect
 	golang.org/x/oauth2 v0.35.0 // indirect
 	golang.org/x/sync v0.21.0 // indirect
 	golang.org/x/sys v0.45.0 // indirect

internal/bus/manager.go

@@ -37,8 +37,10 @@ const (
 	// manager converts messages, and the write queue retains n2k's conservative
 	// default backpressure bound. extraOpts are appended after these defaults so
 	// tests and specialized deployments can override either value.
-	clientReceiveBuffer = 256
-	clientWriteQueue    = 64
+	clientReceiveBuffer  = 256
+	clientWriteQueue     = 64
+	maxCachedDeviceNames = 1024
+	deviceCacheRetention = 6 * time.Hour
 )
 
 var (
@@ -210,9 +212,11 @@ type busClient struct {
 	state   string
 	lastErr error
 
-	supported map[uint64]SupportedPGNs
-	requested map[uint64]time.Time
-	requestWG sync.WaitGroup
+	supported       map[uint64]SupportedPGNs
+	requested       map[uint64]time.Time
+	deviceSeen      map[uint64]time.Time
+	nextDeviceSweep time.Time
+	requestWG       sync.WaitGroup
 }
 
 type busSubscriber struct {
@@ -331,6 +335,9 @@ func (bc *busClient) run(ctx context.Context) {
 		close(iterDone) // retire this iteration's close watchdog
 		bc.mu.Lock()
 		bc.client = nil
+		clear(bc.supported)
+		clear(bc.requested)
+		clear(bc.deviceSeen)
 		bc.mu.Unlock()
 		if ctx.Err() == nil {
 			err := errors.New("receive loop ended; reconnecting")
@@ -479,6 +486,7 @@ func (bc *busClient) recordSupportedPGNs(name uint64, resp *pgn.ParameterGroupNu
 	sort.Slice(values, func(i, j int) bool { return values[i] < values[j] })
 
 	bc.mu.Lock()
+	bc.touchDeviceLocked(name, time.Now())
 	lists := bc.supported[name]
 	switch uint64Value(resp.FunctionCode) {
 	case uint64(pgn.TransmitPGNList):
@@ -499,6 +507,7 @@ func (bc *busClient) maybeRequestPGNList(ctx context.Context, client *n2k.Client
 		return
 	}
 	bc.mu.Lock()
+	bc.touchDeviceLocked(device.RawName, time.Now())
 	if time.Since(bc.requested[device.RawName]) < time.Minute {
 		bc.mu.Unlock()
 		return
@@ -521,6 +530,39 @@ func (bc *busClient) maybeRequestPGNList(ctx context.Context, client *n2k.Client
 	}(device.RawName, source)
 }
 
+// touchDeviceLocked bounds discovery bookkeeping independently of the n2k
+// client's current address registry. Replaced Device NAMEs must not accumulate
+// across a vessel's lifetime. All three maps share the same admission policy.
+func (bc *busClient) touchDeviceLocked(name uint64, now time.Time) {
+	if bc.deviceSeen == nil {
+		bc.deviceSeen = make(map[uint64]time.Time)
+	}
+	remove := func(name uint64) {
+		delete(bc.deviceSeen, name)
+		delete(bc.supported, name)
+		delete(bc.requested, name)
+	}
+	if !now.Before(bc.nextDeviceSweep) {
+		for candidate, seen := range bc.deviceSeen {
+			if now.Sub(seen) >= deviceCacheRetention {
+				remove(candidate)
+			}
+		}
+		bc.nextDeviceSweep = now.Add(time.Minute)
+	}
+	if _, exists := bc.deviceSeen[name]; !exists && len(bc.deviceSeen) >= maxCachedDeviceNames {
+		var oldest uint64
+		var oldestSeen time.Time
+		for candidate, seen := range bc.deviceSeen {
+			if oldestSeen.IsZero() || seen.Before(oldestSeen) {
+				oldest, oldestSeen = candidate, seen
+			}
+		}
+		remove(oldest)
+	}
+	bc.deviceSeen[name] = now
+}
+
 // Devices returns every device currently known across all running bus
 // endpoints, newest activity first. Safe to call concurrently with Acquire/
 // Release; it snapshots the client set under mgr.mu, then reads each client's
@@ -751,6 +793,11 @@ func (h *Handle) Release() {
 			// Acquires of other endpoints. run never takes mgr.mu, so this
 			// cannot deadlock.
 			bc.wg.Wait()
+			mgr.mu.Lock()
+			if mgr.clients[bc.ep] == nil {
+				mgr.met.RemoveSourceDrops("bus:" + bc.ep.Kind + ":" + bc.ep.Name)
+			}
+			mgr.mu.Unlock()
 		}
 	})
 }

internal/bus/manager_test.go

@@ -110,6 +110,29 @@ func TestRecordSupportedPGNsKeepsBothSortedLists(t *testing.T) {
 	}
 }
 
+func TestDiscoveryCachesBoundDeviceNameChurnAndExpireTogether(t *testing.T) {
+	bc := &busClient{supported: map[uint64]SupportedPGNs{}, requested: map[uint64]time.Time{}}
+	tx := uint64(pgn.TransmitPGNList)
+	for name := uint64(1); name <= 10_000; name++ {
+		bc.recordSupportedPGNs(name, &pgn.ParameterGroupNumberListTransmitAndReceive{FunctionCode: &tx})
+		bc.mu.Lock()
+		bc.requested[name] = time.Now()
+		if len(bc.supported) > maxCachedDeviceNames || len(bc.requested) > maxCachedDeviceNames || len(bc.deviceSeen) > maxCachedDeviceNames {
+			t.Fatalf("discovery caches grew beyond %d", maxCachedDeviceNames)
+		}
+		bc.mu.Unlock()
+	}
+	if _, ok := bc.supported[10_000]; !ok {
+		t.Fatal("newest discovery evicted")
+	}
+	bc.mu.Lock()
+	bc.touchDeviceLocked(10_001, time.Now().Add(deviceCacheRetention+time.Minute))
+	bc.mu.Unlock()
+	if len(bc.supported) != 0 || len(bc.requested) != 0 || len(bc.deviceSeen) != 1 {
+		t.Fatalf("expired caches = %d, %d, %d", len(bc.supported), len(bc.requested), len(bc.deviceSeen))
+	}
+}
+
 func TestWriteEncodesToBus(t *testing.T) {
 	fake := busfake.New()
 	m := testManager(t, fake)

internal/connector/connector.go

@@ -180,7 +180,8 @@ func (c *Connector) wake() {
 
 func (c *Connector) intake(ctx context.Context, in <-chan *msg.Envelope, unsub func()) {
 	defer c.wg.Done()
-	defer unsub()
+	unsubscribe := sync.OnceFunc(unsub)
+	defer unsubscribe()
 
 	batch := make([]*msg.Envelope, 0, batchSize)
 	flushTimer := time.NewTicker(batchInterval)
@@ -201,6 +202,7 @@ func (c *Connector) intake(ctx context.Context, in <-chan *msg.Envelope, unsub f
 				c.st.RecordStage(c.cfg.ID, "queued", int64(len(batch)))
 				c.met.ConnectorMessages(context.Background(), c.cfg.ID, "queued", int64(len(batch)))
 				c.recordPrune(context.Background(), pruned)
+				clear(batch)
 				batch = batch[:0]
 				c.wake()
 				return true
@@ -222,68 +224,107 @@ func (c *Connector) intake(ctx context.Context, in <-chan *msg.Envelope, unsub f
 			}
 		}
 	}
-	defer func() {
-		shutdownCtx, cancel := context.WithTimeout(context.Background(), appendTimeout)
-		defer cancel()
-		if !flush(shutdownCtx) && len(batch) > 0 {
-			c.st.RecordStage(c.cfg.ID, "intake_loss", int64(len(batch)))
-			c.met.ConnectorMessages(context.Background(), c.cfg.ID, "intake_loss", int64(len(batch)))
-			c.log.Error("queue append failed during shutdown", "lost", len(batch), "err", shutdownCtx.Err())
+	var interrupted *msg.Envelope
+	ingest := func(ingestCtx context.Context, e *msg.Envelope, alreadyReceived bool) bool {
+		if !alreadyReceived {
+			c.st.RecordConnectorEvent(c.cfg.ID, "received", e)
+			c.met.ConnectorMessages(ctx, c.cfg.ID, "received", 1)
+			c.st.RecordStage(c.cfg.ID, "received", 1)
 		}
-	}()
-
-	ingest := func(e *msg.Envelope) {
-		c.st.RecordConnectorEvent(c.cfg.ID, "received", e)
-		c.met.ConnectorMessages(ctx, c.cfg.ID, "received", 1)
-		c.st.RecordStage(c.cfg.ID, "received", 1)
 		if c.cfg.EffectiveMode() == model.BridgeTransparent && !c.cfg.ForwardManagement && n2kwire.IsManagementPGN(e.PGN) {
 			c.met.ConnectorMessages(ctx, c.cfg.ID, "management_filtered", 1)
 			c.st.RecordStage(c.cfg.ID, "management_filtered", 1)
-			return
+			return true
 		}
-		match, err := c.chain.Match(e)
+		match, err := c.chain.MatchContext(ingestCtx, e)
 		if err != nil {
+			if ingestCtx.Err() != nil {
+				interrupted = e
+				return false
+			}
 			c.met.ConnectorMessages(ctx, c.cfg.ID, "filter_error", 1)
 			c.st.RecordStage(c.cfg.ID, "filter_error", 1)
 			c.log.Debug("filter eval error", "err", err)
-			return
+			return true
 		}
 		if !match {
 			c.met.ConnectorMessages(ctx, c.cfg.ID, "filtered", 1)
 			c.st.RecordStage(c.cfg.ID, "filtered", 1)
-			return
+			return true
 		}
 		c.met.ConnectorMessages(ctx, c.cfg.ID, "matched", 1)
 		batch = append(batch, e)
 		if len(batch) >= batchSize {
-			_ = flush(ctx)
+			return flush(ingestCtx)
 		}
+		return true
 	}
 
-	for {
-		select {
-		case <-ctx.Done():
-			// Drain envelopes already buffered in the subscription so a
-			// graceful stop persists everything the source handed us; the
-			// deferred flush writes whatever remains batched.
-			for {
-				select {
-				case e, ok := <-in:
-					if !ok {
-						return
-					}
-					ingest(e)
-				default:
+	defer func() {
+		// Freeze intake before draining. Hot apply stops the connector while
+		// its source may remain active, so draining a live subscription can
+		// otherwise grow forever. Even an imperfect Runtime unsubscribe cannot
+		// extend this fixed snapshot of outstanding work.
+		unsubscribe()
+		remaining := len(in)
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), appendTimeout)
+		defer cancel()
+		defer func() {
+			lost := len(batch) + remaining
+			if interrupted != nil {
+				lost++
+			}
+			if lost > 0 {
+				c.st.RecordStage(c.cfg.ID, "intake_loss", int64(lost))
+				c.met.ConnectorMessages(context.Background(), c.cfg.ID, "intake_loss", int64(lost))
+				c.log.Error("queue intake incomplete during shutdown", "lost", lost, "err", shutdownCtx.Err())
+			}
+		}()
+		// A canceled normal flush may have left a full batch. Persist it before
+		// admitting anything else, always within the same shutdown deadline.
+		if !flush(shutdownCtx) {
+			return
+		}
+		if interrupted != nil {
+			e := interrupted
+			interrupted = nil
+			if !ingest(shutdownCtx, e, true) {
+				return
+			}
+		}
+		for remaining > 0 {
+			select {
+			case e, ok := <-in:
+				if !ok {
+					remaining = 0
+					break
+				}
+				remaining--
+				if !ingest(shutdownCtx, e, false) {
 					return
 				}
+			default:
+				remaining = 0
 			}
+		}
+		flush(shutdownCtx)
+	}()
+
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		select {
+		case <-ctx.Done():
+			return
 		case <-flushTimer.C:
-			_ = flush(ctx)
+			if !flush(ctx) {
+				return
+			}
 		case e, ok := <-in:
-			if !ok {
+			if !ok || !ingest(ctx, e, false) {
 				return
 			}
-			ingest(e)
 		}
 	}
 }