Cleanups from reviewing #178: single-pass CBOR decode, TTL default, drop metric

deploy/grafana/dashboards/aggregator.json

@@ -545,6 +545,33 @@
           "values": false
         }
       }
+    },
+    {
+      "id": 19,
+      "title": "Commitments Dropped (never reach a block)",
+      "description": "Commitments the service accepted with SUCCESS that are acked out of the queue without being included in a block. 'expired' means the round's reference time reached the request deadline; a backlog exceeding DEFAULT_REQUEST_TTL drops in bulk.",
+      "type": "timeseries",
+      "gridPos": {
+        "h": 8,
+        "w": 12,
+        "x": 0,
+        "y": 60
+      },
+      "targets": [
+        {
+          "expr": "rate(aggregator_commitments_dropped_total[1m])",
+          "legendFormat": "{{instance}} ({{shard}}) {{reason}}/sec",
+          "refId": "A"
+        }
+      ],
+      "fieldConfig": {
+        "defaults": {
+          "unit": "short",
+          "custom": {
+            "fillOpacity": 10
+          }
+        }
+      }
     }
   ]
 }

internal/config/config.go

@@ -123,9 +123,15 @@ type ProcessingConfig struct {
 	SkipDuplicateCheck         bool          `mapstructure:"skip_duplicate_check"`          // Skip finalized record lookup on submit
 }
 
+// DefaultRequestTTLFallback is the lifetime assigned to a request that omits
+// expiresAt. It backs both the DEFAULT_REQUEST_TTL environment default and the
+// zero value, so a config built in code rather than from the environment gets
+// the same TTL the service documents.
+const DefaultRequestTTLFallback = time.Hour
+
 func (c ProcessingConfig) RequestTTL() time.Duration {
 	if c.DefaultRequestTTL == 0 {
-		return time.Hour
+		return DefaultRequestTTLFallback
 	}
 	return c.DefaultRequestTTL
 }
@@ -390,7 +396,7 @@ func Load() (*Config, error) {
 		},
 		Processing: ProcessingConfig{
 			BatchLimit:                 getEnvIntOrDefault("BATCH_LIMIT", 1000),
-			DefaultRequestTTL:          getEnvDurationOrDefault("DEFAULT_REQUEST_TTL", "1h"),
+			DefaultRequestTTL:          getEnvDurationOrDefault("DEFAULT_REQUEST_TTL", DefaultRequestTTLFallback.String()),
 			PrecollectorGracePeriod:    getEnvDurationOrDefault("PRECOLLECTOR_GRACE_PERIOD", "0s"),
 			MaxCommitmentsPerRound:     getEnvIntOrDefault("MAX_COMMITMENTS_PER_ROUND", 20000),
 			CollectPhaseDuration:       getEnvDurationOrDefault("COLLECT_PHASE_DURATION", "200ms"),

internal/metrics/metrics.go

@@ -167,6 +167,26 @@ var (
 		},
 	)
 
+	// CommitmentsDroppedTotal counts drop events for commitments that were
+	// accepted at submit but will never reach a block. Without it a node can
+	// discard an arbitrary volume of acknowledged work with nothing visible on
+	// a dashboard. It counts events rather than distinct commitments: if the
+	// queue ack fails, the commitment is retried and counted again next round.
+	CommitmentsDroppedTotal = promauto.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "aggregator_commitments_dropped_total",
+			Help: "Cumulative drop events for commitments that will never be included in a block.",
+		},
+		[]string{"reason"},
+	)
+
+	// The drop reasons are resolved once, at init. The increments below happen
+	// while roundMutex is held, so they must not pay for a CounterVec label
+	// lookup (an RLock plus a map hash) on every dropped commitment.
+	CommitmentsDroppedExpired   = CommitmentsDroppedTotal.WithLabelValues("expired")
+	CommitmentsDroppedDuplicate = CommitmentsDroppedTotal.WithLabelValues("duplicate")
+	CommitmentsDroppedRejected  = CommitmentsDroppedTotal.WithLabelValues("rejected")
+
 	BFTCertificationDuration = promauto.NewHistogram(
 		prometheus.HistogramOpts{
 			Name:    "aggregator_bft_certification_duration_seconds",

internal/round/batch_processor.go

@@ -43,13 +43,18 @@ func (rm *RoundManager) processMiniBatchForRound(ctx context.Context, round *Rou
 	validCommitments := make([]*models.CertificationRequest, 0, len(commitments))
 	expired := make([]interfaces.CertificationRequestAck, 0)
 	for _, commitment := range commitments {
-		leaf, err := commitmentLeafInput(commitment, round.ReferenceTime)
+		leaf, err := materializeCommitmentLeaf(commitment, round.ReferenceTime)
 		if err != nil {
 			if errors.Is(err, ErrRequestExpired) {
-				rm.logger.WithContext(ctx).Debug("Dropping expired certification request",
+				// Warn, not Debug: this discards work the service already
+				// answered SUCCESS, and a backlog exceeding DEFAULT_REQUEST_TTL
+				// drops requests in bulk.
+				rm.logger.WithContext(ctx).Warn("Dropping expired certification request",
 					"stateID", commitment.StateID.String(),
 					"expiresAt", commitment.CertificationData.ExpiresAt,
+					"effectiveTimeout", commitment.EffectiveTimeout,
 					"referenceTime", round.ReferenceTime)
+				metrics.CommitmentsDroppedExpired.Inc()
 				expired = append(expired, interfaces.CertificationRequestAck{
 					StateID:  commitment.StateID,
 					StreamID: commitment.StreamID,

internal/round/disk_bft_integration_rocksdb_test.go

@@ -347,7 +347,7 @@ func applyMemoryRound(
 	leaves := make([]smtbackend.LeafInput, 0, len(commitments))
 	validCommitments := make([]*models.CertificationRequest, 0, len(commitments))
 	for _, commitment := range commitments {
-		leaf, err := commitmentLeafInput(commitment, 1755000000)
+		leaf, err := materializeCommitmentLeaf(commitment, 1755000000)
 		if err != nil {
 			continue
 		}

internal/round/disk_ha_failover_integration_rocksdb_test.go

@@ -93,7 +93,7 @@ func TestDiskSMTHAFollowerRejectsDivergentFinalizedRoot(t *testing.T) {
 	storage := testutil.SetupTestStorage(t, cfg)
 
 	commitment := testutil.CreateTestCertificationRequest(t, "disk-ha-divergent-root")
-	leaf, err := commitmentLeafInput(commitment, 1755000000)
+	leaf, err := materializeCommitmentLeaf(commitment, 1755000000)
 	require.NoError(t, err)
 
 	wrongRoot := api.NewHexBytes(make([]byte, api.SiblingSize))
@@ -206,7 +206,7 @@ func requirePublishedProof(
 	t.Helper()
 	reader, ok := backend.(smtbackend.PublishedProofReader)
 	require.True(t, ok)
-	leaf, err := commitmentLeafInput(commitment, 1755000000)
+	leaf, err := materializeCommitmentLeaf(commitment, 1755000000)
 	require.NoError(t, err)
 	publishedRoot, err := reader.PublishedRoot(ctx)
 	require.NoError(t, err)

internal/round/leaf_add.go

@@ -32,10 +32,12 @@ func commitmentExpired(commitment *models.CertificationRequest, referenceTime ui
 	return referenceTime >= deadline
 }
 
-// commitmentLeafInput materialises a commitment's SMT leaf under the round's
-// pinned reference time, recording that time on the commitment so the record
-// and the served proof report the value the leaf was actually built from.
-func commitmentLeafInput(commitment *models.CertificationRequest, referenceTime uint64) (smtbackend.LeafInput, error) {
+// materializeCommitmentLeaf materialises a commitment's SMT leaf under the
+// round's pinned reference time. It WRITES commitment.ReferenceTime, so the
+// record and the served proof report the value the leaf was actually built
+// from; the name says materialize rather than build because of that write.
+// models.CertificationRequest.LeafValue is the pure counterpart.
+func materializeCommitmentLeaf(commitment *models.CertificationRequest, referenceTime uint64) (smtbackend.LeafInput, error) {
 	if commitmentExpired(commitment, referenceTime) {
 		return smtbackend.LeafInput{}, ErrRequestExpired
 	}
@@ -85,6 +87,7 @@ func addCommitmentLeaves(
 		if idx < 0 || idx >= len(commitments) {
 			return nil, nil, nil, fmt.Errorf("SMT backend returned invalid duplicate leaf index %d", idx)
 		}
+		metrics.CommitmentsDroppedDuplicate.Inc()
 		dropped = append(dropped, interfaces.CertificationRequestAck{
 			StateID:  commitments[idx].StateID,
 			StreamID: commitments[idx].StreamID,
@@ -102,6 +105,7 @@ func addCommitmentLeaves(
 			"stateID", commitments[rejected.Index].StateID.String(),
 			"reason", string(rejected.Reason),
 			"error", errText)
+		metrics.CommitmentsDroppedRejected.Inc()
 		dropped = append(dropped, interfaces.CertificationRequestAck{
 			StateID:  commitments[rejected.Index].StateID,
 			StreamID: commitments[rejected.Index].StreamID,

internal/round/leaf_add_test.go

@@ -28,11 +28,11 @@ func testCommitment(t *testing.T) *models.CertificationRequest {
 // The leaf a round inserts is built from the round's pinned reference time, and
 // the commitment records that time so the record and the served proof report
 // the value the leaf was actually built from.
-func TestCommitmentLeafInputBindsTheRoundReferenceTime(t *testing.T) {
+func TestMaterializeCommitmentLeafBindsTheRoundReferenceTime(t *testing.T) {
 	const referenceTime uint64 = 1755000000
 	commitment := testCommitment(t)
 
-	leaf, err := commitmentLeafInput(commitment, referenceTime)
+	leaf, err := materializeCommitmentLeaf(commitment, referenceTime)
 	require.NoError(t, err)
 
 	require.Equal(t, referenceTime, commitment.ReferenceTime)
@@ -43,12 +43,12 @@ func TestCommitmentLeafInputBindsTheRoundReferenceTime(t *testing.T) {
 }
 
 // A different round produces a different leaf for the same request.
-func TestCommitmentLeafInputDiffersAcrossRounds(t *testing.T) {
+func TestMaterializeCommitmentLeafDiffersAcrossRounds(t *testing.T) {
 	const referenceTime uint64 = 1755000000
 
-	first, err := commitmentLeafInput(testCommitment(t), referenceTime)
+	first, err := materializeCommitmentLeaf(testCommitment(t), referenceTime)
 	require.NoError(t, err)
-	second, err := commitmentLeafInput(testCommitment(t), referenceTime+1)
+	second, err := materializeCommitmentLeaf(testCommitment(t), referenceTime+1)
 	require.NoError(t, err)
 
 	require.Equal(t, first.Key, second.Key)
@@ -61,28 +61,28 @@ func TestServiceAssignedDeadlineStillBindsReferenceTime(t *testing.T) {
 	commitment.CertificationData.ExpiresAt = nil
 	commitment.EffectiveTimeout = referenceTime + 3600
 
-	leaf, err := commitmentLeafInput(commitment, referenceTime)
+	leaf, err := materializeCommitmentLeaf(commitment, referenceTime)
 	require.NoError(t, err)
 	require.Equal(t, api.LeafValue(commitment.CertificationData.TransactionHash.DataBytes(), referenceTime), leaf.Value)
 	require.NotEqual(t, commitment.CertificationData.TransactionHash.DataBytes(), leaf.Value)
 
-	_, err = commitmentLeafInput(commitment, commitment.EffectiveTimeout)
+	_, err = materializeCommitmentLeaf(commitment, commitment.EffectiveTimeout)
 	require.ErrorIs(t, err, ErrRequestExpired)
 }
 
 // A request may only be inserted in a round whose reference time is strictly
 // below its timeout; an expired one is reported so it can be acked out of the
 // queue rather than retried forever.
-func TestCommitmentLeafInputRejectsAnExpiredRequest(t *testing.T) {
+func TestMaterializeCommitmentLeafRejectsAnExpiredRequest(t *testing.T) {
 	commitment := testCommitment(t)
 
-	_, err := commitmentLeafInput(commitment, testExpiresAt-1)
+	_, err := materializeCommitmentLeaf(commitment, testExpiresAt-1)
 	require.NoError(t, err)
 
-	_, err = commitmentLeafInput(commitment, testExpiresAt)
+	_, err = materializeCommitmentLeaf(commitment, testExpiresAt)
 	require.ErrorIs(t, err, ErrRequestExpired)
 
-	_, err = commitmentLeafInput(commitment, testExpiresAt+1)
+	_, err = materializeCommitmentLeaf(commitment, testExpiresAt+1)
 	require.ErrorIs(t, err, ErrRequestExpired)
 }
 

internal/round/precollector.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/unicitynetwork/aggregator-go/internal/logger"
+	"github.com/unicitynetwork/aggregator-go/internal/metrics"
 	"github.com/unicitynetwork/aggregator-go/internal/models"
 	smtbackend "github.com/unicitynetwork/aggregator-go/internal/smt/backend"
 	"github.com/unicitynetwork/aggregator-go/internal/storage/interfaces"
@@ -341,13 +342,16 @@ func (cp *childPrecollector) addBatch(
 	expired := make([]interfaces.CertificationRequestAck, 0)
 
 	for _, c := range commitments {
-		leaf, err := commitmentLeafInput(c, referenceTime)
+		leaf, err := materializeCommitmentLeaf(c, referenceTime)
 		if err != nil {
 			if errors.Is(err, ErrRequestExpired) {
-				cp.logger.WithContext(ctx).Debug("Dropping expired certification request",
+				// See the matching drop in batch_processor.go: Warn, not Debug.
+				cp.logger.WithContext(ctx).Warn("Dropping expired certification request",
 					"stateID", c.StateID.String(),
 					"expiresAt", c.CertificationData.ExpiresAt,
+					"effectiveTimeout", c.EffectiveTimeout,
 					"referenceTime", referenceTime)
+				metrics.CommitmentsDroppedExpired.Inc()
 				expired = append(expired, interfaces.CertificationRequestAck{
 					StateID:  c.StateID,
 					StreamID: c.StreamID,

internal/round/round_process_regression_test.go

@@ -131,7 +131,7 @@ func TestRoundProcessingUsesScheduledRoundSnapshot(t *testing.T) {
 	}, time.Second, 10*time.Millisecond)
 
 	roundTwoCommitment := testutil.CreateTestCertificationRequest(t, "scheduled_round_two")
-	roundTwoLeaf, err := commitmentLeafInput(roundTwoCommitment, 1755000000)
+	roundTwoLeaf, err := materializeCommitmentLeaf(roundTwoCommitment, 1755000000)
 	require.NoError(t, err)
 	roundTwoSnapshot, err := rm.smtBackend.CreateSnapshot(ctx)
 	require.NoError(t, err)
@@ -372,7 +372,7 @@ func TestStartNewRoundWithSnapshotAbandonAfterSnapshotCleanupResetsRedisPendingS
 	rm.markProofsPending([]*models.CertificationRequest{oldCommitment})
 
 	newCommitment := testutil.CreateTestCertificationRequest(t, "new_precollected_round")
-	newLeaf, err := commitmentLeafInput(newCommitment, 1755000000)
+	newLeaf, err := materializeCommitmentLeaf(newCommitment, 1755000000)
 	require.NoError(t, err)
 	newSnapshot, err := rm.smtBackend.CreateSnapshot(ctx)
 	require.NoError(t, err)
@@ -457,7 +457,7 @@ func TestStartNewRoundWithSnapshotDoesNotReplayFinalizedRoundHistory(t *testing.
 	}
 
 	newCommitment := testutil.CreateTestCertificationRequest(t, "precollected_round_pending_marker")
-	newLeaf, err := commitmentLeafInput(newCommitment, 1755000000)
+	newLeaf, err := materializeCommitmentLeaf(newCommitment, 1755000000)
 	require.NoError(t, err)
 	newSnapshot, err := rm.smtBackend.CreateSnapshot(ctx)
 	require.NoError(t, err)
@@ -521,7 +521,7 @@ func TestStaleCertificationRequestAbandonsStoredDurableProposal(t *testing.T) {
 	}()
 
 	commitment := testutil.CreateTestCertificationRequest(t, "stale_durable_proposal")
-	leaf, err := commitmentLeafInput(commitment, 1755000000)
+	leaf, err := materializeCommitmentLeaf(commitment, 1755000000)
 	require.NoError(t, err)
 	snapshot, err := rm.smtBackend.CreateSnapshot(ctx)
 	require.NoError(t, err)
@@ -598,7 +598,7 @@ func TestStartNewRoundRetriesEqualFinalizingRoundProposal(t *testing.T) {
 	}()
 
 	commitment := testutil.CreateTestCertificationRequest(t, "repeat_uc_equal_round_retry")
-	leaf, err := commitmentLeafInput(commitment, 1755000000)
+	leaf, err := materializeCommitmentLeaf(commitment, 1755000000)
 	require.NoError(t, err)
 	snapshot := testRMSnapshot(t, ctx, rm)
 	result, err := snapshot.AddLeavesClassified(ctx, []smtbackend.LeafInput{leaf})