diff --git a/core/services/llo/telem/telemetry.go b/core/services/llo/telem/telemetry.go index 572427659e4..ba1906a0481 100644 --- a/core/services/llo/telem/telemetry.go +++ b/core/services/llo/telem/telemetry.go @@ -26,6 +26,14 @@ import ( const adapterLWBAErrorName = "AdapterLWBAError" +// maxBufferedSeqNrsPerDigest bounds telemetryBuffer independently of +// Transmit() activity. Buffered entries are normally evicted by +// sendBufferedTelemetry when a Transmit() occurs for their digest, but a +// digest that transmits rarely (or never, e.g. because it lost transmission +// duty or is winding down) would otherwise accumulate one entry per round +// forever. +const maxBufferedSeqNrsPerDigest = 10 + // DSOpts is the shared, version-agnostic LLO data-source options (llo/v30 and // llo/v31 both use llodatasource.DSOpts). Aliased here so the telemetry and // observation paths keep referring to telem.DSOpts. @@ -143,7 +151,8 @@ type telemeter struct { } // Buffer Report and Outcome telemetry to only send - // for transmitting rounds sequence numbers + // for transmitting rounds sequence numbers. Bounded per-digest by + // maxBufferedSeqNrsPerDigest; see evictOldestSeqNrsLocked. telemetryBufferMu sync.Mutex telemetryBuffer map[string]map[uint64][]telemetryEntry @@ -319,6 +328,26 @@ func (t *telemeter) sendBufferedTelemetry(digest types.ConfigDigest, seqNr uint6 }() } +// evictOldestSeqNrsLocked drops the oldest buffered seqNrs for digest until +// at most maxBufferedSeqNrsPerDigest remain. Callers must hold +// telemetryBufferMu. +func (t *telemeter) evictOldestSeqNrsLocked(digest string) { + digestMessages := t.telemetryBuffer[digest] + for len(digestMessages) > maxBufferedSeqNrsPerDigest { + var oldest uint64 + first := true + for seqNr := range digestMessages { + if first || seqNr < oldest { + oldest = seqNr + first = false + } + } + delete(digestMessages, oldest) + t.eng.Warnw("Telemetry: evicted buffered telemetry for stale seqNr; digest may not be transmitting", + "digest", digest, "evictedSeqNr", oldest, "bufferSize", len(digestMessages)) + } +} + func (t *telemeter) enqueueTelemetry(digest string, seqNr uint64, typ synchronization.TelemetryType, msg proto.Message) { switch typ { case synchronization.PipelineBridge, synchronization.LLOObservation, synchronization.EnhancedEAMercury: @@ -349,6 +378,7 @@ func (t *telemeter) enqueueTelemetry(digest string, seqNr uint64, typ synchroniz telemType: typ, msg: msg, }} + t.evictOldestSeqNrsLocked(digest) default: // synchronization.LLOReport and other buffered types // Report telemetry: append, since multiple reports per seqNr is // expected (one per reportable channel). @@ -363,6 +393,7 @@ func (t *telemeter) enqueueTelemetry(digest string, seqNr uint64, typ synchroniz telemType: typ, msg: msg, }) + t.evictOldestSeqNrsLocked(digest) } } diff --git a/core/services/llo/telem/telemetry_test.go b/core/services/llo/telem/telemetry_test.go index 01231f6ce90..88d02971de5 100644 --- a/core/services/llo/telem/telemetry_test.go +++ b/core/services/llo/telem/telemetry_test.go @@ -824,7 +824,7 @@ func Test_Telemeter_outcomeTelemetry_samplingAtFlushTime(t *testing.T) { // second transmits — mimicking a DON where DeltaRound << report interval. const ( secondsCovered = 3 - outcomesPerSecond = 5 + outcomesPerSecond = 3 // secondsCovered*outcomesPerSecond must stay <= maxBufferedSeqNrsPerDigest baseObservationUnix = int64(1737936858) baseSeqNr = uint64(1000) ) @@ -981,7 +981,7 @@ func Test_Telemeter_reportTelemetry_samplingAtFlushTime(t *testing.T) { const ( secondsCovered = 3 - seqNrsPerSecond = 5 + seqNrsPerSecond = 3 // secondsCovered*seqNrsPerSecond must stay <= maxBufferedSeqNrsPerDigest baseObservationUnix = int64(1737936858) baseSeqNr = uint64(2000) ) @@ -1193,3 +1193,67 @@ func Test_Telemeter_reportTelemetry_samplingAtFlushTime(t *testing.T) { "each per-channel report should be admitted (distinct sampler fingerprints)") }) } + +// Test_Telemeter_telemetryBuffer_boundedWithoutTransmit guards against a +// digest that never (or rarely) transmits accumulating an unbounded number of +// buffered seqNrs. Previously telemetryBuffer entries were only evicted by +// sendBufferedTelemetry on TrackSeqNr, so a stuck/inactive digest would grow +// the buffer forever. +func Test_Telemeter_telemetryBuffer_boundedWithoutTransmit(t *testing.T) { + t.Parallel() + + lggr := logger.TestLogger(t) + donID := uint32(1) + cd := (&mockOpts{}).ConfigDigest() + + t.Run("outcome telemetry", func(t *testing.T) { + t.Parallel() + m := &mockMonitoringEndpoint{chTypedLogs: make(chan typedLog, 100)} + tm := newTelemeter(TelemeterParams{ + Logger: lggr, + MonitoringEndpoint: m, + DonID: donID, + CaptureOutcomeTelemetry: true, + }) + + const rounds = maxBufferedSeqNrsPerDigest * 5 + for i := range rounds { + seqNr := uint64(i) + tm.enqueueTelemetry(cd.Hex(), seqNr, synchronization.LLOOutcome, &lloprotocol.LLOOutcomeTelemetry{ + SeqNr: seqNr, + ConfigDigest: cd[:], + }) + } + + tm.telemetryBufferMu.Lock() + defer tm.telemetryBufferMu.Unlock() + assert.LessOrEqual(t, len(tm.telemetryBuffer[cd.Hex()]), maxBufferedSeqNrsPerDigest) + // the most recent seqNr must always survive eviction + assert.Contains(t, tm.telemetryBuffer[cd.Hex()], uint64(rounds-1)) + }) + + t.Run("report telemetry", func(t *testing.T) { + t.Parallel() + m := &mockMonitoringEndpoint{chTypedLogs: make(chan typedLog, 100)} + tm := newTelemeter(TelemeterParams{ + Logger: lggr, + MonitoringEndpoint: m, + DonID: donID, + CaptureReportTelemetry: true, + }) + + const rounds = maxBufferedSeqNrsPerDigest * 5 + for i := range rounds { + seqNr := uint64(i) + tm.enqueueTelemetry(cd.Hex(), seqNr, synchronization.LLOReport, &lloprotocol.LLOReportTelemetry{ + SeqNr: seqNr, + ConfigDigest: cd[:], + }) + } + + tm.telemetryBufferMu.Lock() + defer tm.telemetryBufferMu.Unlock() + assert.LessOrEqual(t, len(tm.telemetryBuffer[cd.Hex()]), maxBufferedSeqNrsPerDigest) + assert.Contains(t, tm.telemetryBuffer[cd.Hex()], uint64(rounds-1)) + }) +}