Durable Emitter - TTL budget metric

pkg/durableemitter/durable_emitter.go

@@ -780,14 +780,6 @@ func (d *DurableEmitter) expiryLoop() {
 	}
 }
 
-func (d *DurableEmitter) queueStatsNearExpiryLead() time.Duration {
-	lead := 5 * time.Minute
-	if d.cfg.Metrics != nil && d.cfg.Metrics.NearExpiryLead > 0 {
-		lead = d.cfg.Metrics.NearExpiryLead
-	}
-	return lead
-}
-
 func (d *DurableEmitter) metricsLoop() {
 	mc := d.cfg.Metrics
 	poll := mc.PollInterval
@@ -806,7 +798,7 @@ func (d *DurableEmitter) metricsLoop() {
 			return
 		case <-ticker.C:
 			if obs, ok := d.store.(DurableQueueObserver); ok {
-				st, err := obs.ObserveDurableQueue(ctx, d.cfg.EventTTL, d.queueStatsNearExpiryLead())
+				st, err := obs.ObserveDurableQueue(ctx, d.cfg.EventTTL)
 				if err != nil {
 					d.eng.Debugw("DurableEmitter: queue observe failed; keeping last depth", "error", err)
 				} else {

pkg/durableemitter/durable_emitter_metrics.go

@@ -20,8 +20,6 @@ import (
 type DurableEmitterMetricsConfig struct {
 	// PollInterval is how often queue and optional process gauges refresh. Zero = 10s.
 	PollInterval time.Duration
-	// NearExpiryLead is the window before EventTTL used for queue.near_ttl (DLQ pressure proxy). Zero = 5m.
-	NearExpiryLead time.Duration
 	// MaxQueuePayloadBytes, if > 0, records capacity_usage_ratio = queue_payload_bytes / max.
 	MaxQueuePayloadBytes int64
 }
@@ -64,7 +62,7 @@ type durableEmitterMetrics struct {
 	queueDepth         metric.Int64Gauge
 	queuePayloadBytes  metric.Int64Gauge
 	queueOldestAgeSec  metric.Float64Gauge
-	queueNearTTL       metric.Int64Gauge
+	queueTTLBudgetSec  metric.Float64Gauge
 	queueCapacityRatio metric.Float64Gauge
 	procHeapInuse      metric.Int64Gauge
 	procHeapSys        metric.Int64Gauge
@@ -230,10 +228,10 @@ func newDurableEmitterMetrics(meter metric.Meter) (*durableEmitterMetrics, error
 	); err != nil {
 		return nil, err
 	}
-	if m.queueNearTTL, err = meter.Int64Gauge(
-		"durable_emitter.queue.near_ttl",
-		metric.WithUnit("{row}"),
-		metric.WithDescription("Rows within near-expiry window of EventTTL (DLQ pressure proxy; no separate DLQ table)"),
+	if m.queueTTLBudgetSec, err = meter.Float64Gauge(
+		"durable_emitter.queue.ttl_budget_seconds",
+		metric.WithUnit("s"),
+		metric.WithDescription("Seconds of TTL headroom for the oldest pending event (EventTTL - oldest age); low/negative → DLQ/expiry pressure. Alert engine decides what 'near' means"),
 	); err != nil {
 		return nil, err
 	}
@@ -309,7 +307,7 @@ func (m *durableEmitterMetrics) recordStoreOp(ctx context.Context, op string, el
 }
 
 // recordQueueStats records the DB-derived queue statistics (payload bytes,
-// oldest pending age, near-TTL count) from an already-observed snapshot. The
+// oldest pending age, TTL budget) from an already-observed snapshot. The
 // queue depth gauge itself is recorded separately by DurableEmitter from the
 // same snapshot's authoritative TotalRows count.
 func (m *durableEmitterMetrics) recordQueueStats(ctx context.Context, st DurableQueueStats, maxBytes int64) {
@@ -322,7 +320,7 @@ func (m *durableEmitterMetrics) recordQueueStats(ctx context.Context, st Durable
 	} else {
 		m.queueOldestAgeSec.Record(ctx, st.OldestPendingAge.Seconds())
 	}
-	m.queueNearTTL.Record(ctx, st.NearTTLCount)
+	m.queueTTLBudgetSec.Record(ctx, st.TTLBudget.Seconds())
 	if maxBytes > 0 {
 		m.queueCapacityRatio.Record(ctx, float64(st.PayloadBytes)/float64(maxBytes))
 	}

pkg/durableemitter/durable_emitter_test.go

@@ -1460,12 +1460,13 @@ func (m *MemDurableEventStore) Len() int {
 }
 
 // ObserveDurableQueue implements DurableQueueObserver.
-func (m *MemDurableEventStore) ObserveDurableQueue(_ context.Context, eventTTL, nearExpiryLead time.Duration) (DurableQueueStats, error) {
+func (m *MemDurableEventStore) ObserveDurableQueue(_ context.Context, eventTTL time.Duration) (DurableQueueStats, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	now := time.Now()
 	var st DurableQueueStats
 	st.TotalRows = int64(len(m.events))
+	st.TTLBudget = eventTTL
 	if len(m.events) == 0 {
 		return st, nil
 	}
@@ -1478,14 +1479,8 @@ func (m *MemDurableEventStore) ObserveDurableQueue(_ context.Context, eventTTL,
 			oldest = e.CreatedAt
 			first = false
 		}
-		age := now.Sub(e.CreatedAt)
-		if eventTTL > 0 && nearExpiryLead > 0 && nearExpiryLead < eventTTL {
-			threshold := eventTTL - nearExpiryLead
-			if age >= threshold && age < eventTTL {
-				st.NearTTLCount++
-			}
-		}
 	}
 	st.OldestPendingAge = now.Sub(oldest)
+	st.TTLBudget = eventTTL - st.OldestPendingAge
 	return st, nil
 }

pkg/durableemitter/observable_store.go

@@ -26,18 +26,20 @@ type DurableQueueStats struct {
 	TotalRows        int64
 	PayloadBytes     int64
 	OldestPendingAge time.Duration // 0 if the queue is empty
-	// NearTTLCount is the number of rows within nearExpiryLead of EventTTL (still
-	// pending, not yet removed by expiry). Serves as a DLQ-pressure proxy; there is
-	// no separate dead-letter table in the default design.
-	NearTTLCount int64
+	// TTLBudget is the remaining time before the oldest still-pending event hits
+	// EventTTL (i.e. EventTTL - OldestPendingAge). It is the headroom of the event
+	// closest to expiry and serves as a DLQ-pressure proxy; there is no separate
+	// dead-letter table in the default design. It goes negative when the expiry
+	// loop is behind. Equals EventTTL when the queue is empty.
+	TTLBudget time.Duration
 }
 
 // DurableQueueObserver is optionally implemented by DurableEventStore implementations
 // so DurableEmitter can export queue depth and age gauges when metrics are enabled.
 type DurableQueueObserver interface {
-	// ObserveDurableQueue returns live queue statistics. eventTTL and nearExpiryLead
-	// match Config (nearExpiryLead should be << eventTTL).
-	ObserveDurableQueue(ctx context.Context, eventTTL, nearExpiryLead time.Duration) (DurableQueueStats, error)
+	// ObserveDurableQueue returns live queue statistics. eventTTL matches Config and
+	// is used to derive the remaining TTL budget of the oldest pending event.
+	ObserveDurableQueue(ctx context.Context, eventTTL time.Duration) (DurableQueueStats, error)
 }
 
 // BatchInserter is optionally implemented by DurableEventStore implementations
@@ -125,12 +127,12 @@ func (s *metricsInstrumentedStore) DeleteExpired(ctx context.Context, ttl time.D
 	return n, err
 }
 
-func (s *metricsInstrumentedStore) ObserveDurableQueue(ctx context.Context, eventTTL, nearExpiryLead time.Duration) (DurableQueueStats, error) {
+func (s *metricsInstrumentedStore) ObserveDurableQueue(ctx context.Context, eventTTL time.Duration) (DurableQueueStats, error) {
 	o, ok := s.inner.(DurableQueueObserver)
 	if !ok {
 		return DurableQueueStats{}, errors.New("inner DurableEventStore does not implement DurableQueueObserver")
 	}
-	return o.ObserveDurableQueue(ctx, eventTTL, nearExpiryLead)
+	return o.ObserveDurableQueue(ctx, eventTTL)
 }
 
 func (s *metricsInstrumentedStore) InsertBatch(ctx context.Context, payloads [][]byte) ([]int64, error) {

pkg/durableemitter/store.go

@@ -165,7 +165,7 @@ type chipDurableQueueAgg struct {
 // cnt/payload_sum/min_created describe only the undelivered backlog, while total
 // is the authoritative physical row count (incl. delivered-but-not-purged rows)
 // used as the queue-depth gauge.
-func (s *PgDurableEventStore) ObserveDurableQueue(ctx context.Context, eventTTL, nearExpiryLead time.Duration) (DurableQueueStats, error) {
+func (s *PgDurableEventStore) ObserveDurableQueue(ctx context.Context, eventTTL time.Duration) (DurableQueueStats, error) {
 	const qAgg = `
 SELECT
 	count(*) FILTER (WHERE delivered_at IS NULL)::bigint AS cnt,
@@ -182,21 +182,13 @@ FROM ` + chipDurableEventsTable
 	st.Depth = row.Cnt
 	st.TotalRows = row.Total
 	st.PayloadBytes = row.PayloadSum
+	// TTLBudget is the headroom of the oldest pending event before it hits EventTTL.
+	// With no backlog the whole budget is available; the "near expiry" threshold is
+	// applied downstream by the alert engine, not here.
+	st.TTLBudget = eventTTL
 	if row.MinCreated != nil {
 		st.OldestPendingAge = time.Since(*row.MinCreated)
-	}
-	if eventTTL > 0 && nearExpiryLead > 0 && nearExpiryLead < eventTTL {
-		ttlSec := int64(eventTTL.Round(time.Second) / time.Second)
-		leadSec := int64(nearExpiryLead.Round(time.Second) / time.Second)
-		const qNear = `
-SELECT count(*)::bigint
-FROM ` + chipDurableEventsTable + `
-WHERE delivered_at IS NULL
-  AND created_at >= now() - ($1::bigint * interval '1 second')
-  AND created_at < now() - (($1::bigint - $2::bigint) * interval '1 second')`
-		if err := s.ds.GetContext(ctx, &st.NearTTLCount, qNear, ttlSec, leadSec); err != nil {
-			return DurableQueueStats{}, fmt.Errorf("durable queue near-ttl: %w", err)
-		}
+		st.TTLBudget = eventTTL - st.OldestPendingAge
 	}
 	return st, nil
 }