From c8196455b2db5b6266e67873113c7be5acc33bb8 Mon Sep 17 00:00:00 2001 From: Pavel Grigorenko Date: Mon, 24 Aug 2026 20:47:59 +0200 Subject: [PATCH 1/4] perf(api): decode CertificationData once instead of probing first UnmarshalCBOR ran UnmarshalTagged to check the tag and element count, then UnmarshalTaggedValue to decode the same bytes again. The probe pass was redundant: the toarray decode rejects a wrong tag and a wrong element count on its own, so only the version check needs the decoded value. This is the per-request path (internal/gateway/handlers.go decodes every certification_request), and the probe was roughly half of the decode cost: before 5701-5781 ns/op 1288 B/op 32 allocs/op after 3010-3098 ns/op 576 B/op 13 allocs/op --- pkg/api/certification_data_decode_test.go | 129 ++++++++++++++++++++++ pkg/api/certification_request.go | 29 ++--- 2 files changed, 138 insertions(+), 20 deletions(-) create mode 100644 pkg/api/certification_data_decode_test.go diff --git a/pkg/api/certification_data_decode_test.go b/pkg/api/certification_data_decode_test.go new file mode 100644 index 0000000..9c7cc14 --- /dev/null +++ b/pkg/api/certification_data_decode_test.go @@ -0,0 +1,129 @@ +package api + +import ( + "testing" + + "github.com/stretchr/testify/require" + "github.com/unicitynetwork/bft-go-base/types" +) + +// CertificationData.UnmarshalCBOR relies on the toarray decode to reject a +// wrong tag and a wrong element count, rather than probing the payload first. +// These cases pin that validation surface so it cannot be weakened silently. +func TestCertificationDataUnmarshalValidationSurface(t *testing.T) { + type sixField struct { + _ struct{} `cbor:",toarray"` + Version types.Version + OwnerPredicate Predicate + SourceStateHash SourceStateHash + TransactionHash TransactionHash + ExpiresAt *uint64 + Witness HexBytes + } + type fiveField struct { + _ struct{} `cbor:",toarray"` + Version types.Version + OwnerPredicate Predicate + SourceStateHash SourceStateHash + TransactionHash TransactionHash + Witness HexBytes + } + type sevenField struct { + _ struct{} `cbor:",toarray"` + Version types.Version + OwnerPredicate Predicate + SourceStateHash SourceStateHash + TransactionHash TransactionHash + ExpiresAt *uint64 + Witness HexBytes + Extra uint64 + } + + predicate := Predicate{Engine: 1, Code: []byte{0x01}, Params: []byte{0x02}} + sourceStateHash := RequireNewImprintV2("cd60000000000000000000000000000000000000000000000000000000000000") + transactionHash := RequireNewImprintV2("cd61000000000000000000000000000000000000000000000000000000000000") + witness := HexBytes(make([]byte, 65)) + + base := sixField{ + Version: CertificationDataVersion, OwnerPredicate: predicate, + SourceStateHash: sourceStateHash, TransactionHash: transactionHash, + ExpiresAt: Uint64Ptr(1755003600), Witness: witness, + } + marshal := func(tag uint64, v any) []byte { + b, err := types.Cbor.MarshalTaggedValue(tag, v) + require.NoError(t, err) + return b + } + + nilExpiry := base + nilExpiry.ExpiresAt = nil + v1 := base + v1.Version = 1 + v3 := base + v3.Version = 3 + v0 := base + v0.Version = 0 + + tests := []struct { + name string + data []byte + accept bool + }{ + {"explicit ExpiresAt", marshal(CertificationDataTag, &base), true}, + {"absent ExpiresAt", marshal(CertificationDataTag, &nilExpiry), true}, + {"version 0", marshal(CertificationDataTag, &v0), false}, + {"version 1", marshal(CertificationDataTag, &v1), false}, + {"version 3", marshal(CertificationDataTag, &v3), false}, + {"wrong tag", marshal(CertificationRequestTag, &base), false}, + {"too few fields", marshal(CertificationDataTag, &fiveField{ + Version: CertificationDataVersion, OwnerPredicate: predicate, + SourceStateHash: sourceStateHash, TransactionHash: transactionHash, Witness: witness, + }), false}, + {"too many fields", marshal(CertificationDataTag, &sevenField{ + Version: CertificationDataVersion, OwnerPredicate: predicate, + SourceStateHash: sourceStateHash, TransactionHash: transactionHash, + ExpiresAt: Uint64Ptr(1755003600), Witness: witness, Extra: 9, + }), false}, + {"tagged non-array", marshal(CertificationDataTag, uint64(7)), false}, + {"empty", []byte{}, false}, + {"garbage", []byte{0xff, 0xff, 0xff}, false}, + {"truncated", marshal(CertificationDataTag, &base)[:5], false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var cd CertificationData + err := cd.UnmarshalCBOR(tt.data) + if tt.accept { + require.NoError(t, err) + require.Equal(t, CertificationDataVersion, cd.Version) + return + } + require.Error(t, err) + }) + } +} + +// An absent deadline must survive a decode/encode round trip as absent, not as +// zero: zero is a legal instant, so the two cannot share a representation. +func TestCertificationDataAbsentExpiresAtRoundTrips(t *testing.T) { + original := &CertificationData{ + Version: CertificationDataVersion, + OwnerPredicate: Predicate{Engine: 1, Code: []byte{0x01}, Params: []byte{0x02}}, + SourceStateHash: RequireNewImprintV2("cd60000000000000000000000000000000000000000000000000000000000000"), + TransactionHash: RequireNewImprintV2("cd61000000000000000000000000000000000000000000000000000000000000"), + Witness: make([]byte, 65), + } + encoded, err := original.MarshalCBOR() + require.NoError(t, err) + + var decoded CertificationData + require.NoError(t, decoded.UnmarshalCBOR(encoded)) + require.Nil(t, decoded.ExpiresAt) + + zero := *original + zero.ExpiresAt = Uint64Ptr(0) + zeroEncoded, err := zero.MarshalCBOR() + require.NoError(t, err) + require.NotEqual(t, encoded, zeroEncoded, "absent and zero must not encode identically") +} diff --git a/pkg/api/certification_request.go b/pkg/api/certification_request.go index 0d86dd5..56562b8 100644 --- a/pkg/api/certification_request.go +++ b/pkg/api/certification_request.go @@ -176,30 +176,19 @@ func (c *CertificationData) MarshalCBOR() ([]byte, error) { } func (c *CertificationData) UnmarshalCBOR(data []byte) error { - tag, arr, err := types.Cbor.UnmarshalTagged(data) - if err != nil { - return err - } - if tag != CertificationDataTag { - return errors.New("invalid CertificationData tag") - } // One version, one element count. ExpiresAt holds its position even when it - // carries no value, so the array length never depends on the payload. - if len(arr) != certificationDataFieldCount { - return fmt.Errorf("CertificationData: expected %d fields, got %d", - certificationDataFieldCount, len(arr)) - } - version, ok := arr[0].(uint64) - if !ok { - return errors.New("invalid CertificationData version") - } - if types.Version(version) != CertificationDataVersion { - return fmt.Errorf("unsupported CertificationData version: %d", version) - } + // carries no value, so the array length never depends on the payload, and + // the toarray decode below rejects a wrong tag or a wrong element count on + // its own. Decoding once and checking the version off the result costs + // about half of what a separate tag-and-length probe pass did, on the + // per-request path. type alias CertificationData var decoded alias if err := types.Cbor.UnmarshalTaggedValue(CertificationDataTag, data, &decoded); err != nil { - return err + return fmt.Errorf("CertificationData: %w", err) + } + if decoded.Version != CertificationDataVersion { + return fmt.Errorf("unsupported CertificationData version: %d", decoded.Version) } *c = CertificationData(decoded) return nil From 5665511005d00b9b92d3152ee94d68065668728d Mon Sep 17 00:00:00 2001 From: Pavel Grigorenko Date: Mon, 24 Aug 2026 20:48:46 +0200 Subject: [PATCH 2/4] refactor(config): single source for the default request TTL The 1h default was written twice: once as the DEFAULT_REQUEST_TTL environment default and once as the zero-value fallback in RequestTTL(). The two could drift, and which one applied depended on whether the config came from the environment or was built in code. Both now derive from DefaultRequestTTLFallback. Validate still accepts 0, which remains the "unset, use the default" signal for programmatic configs. --- internal/config/config.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/internal/config/config.go b/internal/config/config.go index 017eaf5..173e421 100644 --- a/internal/config/config.go +++ b/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"), From ffb3e317a7675b6b42ed06b3512d896358d59958 Mon Sep 17 00:00:00 2001 From: Pavel Grigorenko Date: Mon, 24 Aug 2026 20:49:32 +0200 Subject: [PATCH 3/4] refactor(round): name the leaf builder for the write it performs commitmentLeafInput reads as a pure builder but assigns commitment.ReferenceTime. The adjacent models.CertificationRequest.LeafValue is genuinely pure, and the PR's own test asserts that -- so the two sat side by side with opposite contracts and near-identical names. Renamed to materializeCommitmentLeaf and made the write explicit in the doc comment. No behaviour change. --- internal/round/batch_processor.go | 2 +- .../round/disk_bft_integration_rocksdb_test.go | 2 +- .../disk_ha_failover_integration_rocksdb_test.go | 4 ++-- internal/round/leaf_add.go | 10 ++++++---- internal/round/leaf_add_test.go | 16 ++++++++-------- internal/round/precollector.go | 2 +- internal/round/round_process_regression_test.go | 10 +++++----- 7 files changed, 24 insertions(+), 22 deletions(-) diff --git a/internal/round/batch_processor.go b/internal/round/batch_processor.go index c3ebcda..c45f18a 100644 --- a/internal/round/batch_processor.go +++ b/internal/round/batch_processor.go @@ -43,7 +43,7 @@ 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", diff --git a/internal/round/disk_bft_integration_rocksdb_test.go b/internal/round/disk_bft_integration_rocksdb_test.go index d57c402..a957fbf 100644 --- a/internal/round/disk_bft_integration_rocksdb_test.go +++ b/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 } diff --git a/internal/round/disk_ha_failover_integration_rocksdb_test.go b/internal/round/disk_ha_failover_integration_rocksdb_test.go index a51dd47..88a3ab0 100644 --- a/internal/round/disk_ha_failover_integration_rocksdb_test.go +++ b/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) diff --git a/internal/round/leaf_add.go b/internal/round/leaf_add.go index 3b4a90b..8e8d339 100644 --- a/internal/round/leaf_add.go +++ b/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 } diff --git a/internal/round/leaf_add_test.go b/internal/round/leaf_add_test.go index 398fd52..cc8633c 100644 --- a/internal/round/leaf_add_test.go +++ b/internal/round/leaf_add_test.go @@ -32,7 +32,7 @@ func TestCommitmentLeafInputBindsTheRoundReferenceTime(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) @@ -46,9 +46,9 @@ func TestCommitmentLeafInputBindsTheRoundReferenceTime(t *testing.T) { func TestCommitmentLeafInputDiffersAcrossRounds(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,12 +61,12 @@ 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) } @@ -76,13 +76,13 @@ func TestServiceAssignedDeadlineStillBindsReferenceTime(t *testing.T) { func TestCommitmentLeafInputRejectsAnExpiredRequest(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) } diff --git a/internal/round/precollector.go b/internal/round/precollector.go index c2922c2..9b1d499 100644 --- a/internal/round/precollector.go +++ b/internal/round/precollector.go @@ -341,7 +341,7 @@ 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", diff --git a/internal/round/round_process_regression_test.go b/internal/round/round_process_regression_test.go index 258d3a4..3987d06 100644 --- a/internal/round/round_process_regression_test.go +++ b/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}) From 93dab79a2a9fb7b29632b73624c61850fb18ad6f Mon Sep 17 00:00:00 2001 From: Pavel Grigorenko Date: Mon, 24 Aug 2026 20:51:29 +0200 Subject: [PATCH 4/4] feat(metrics): count commitments dropped without reaching a block An expired request is acked out of the queue after the service already answered SUCCESS, leaving no aggregator record and no durable trace. It was logged at Debug with no counter, so a node could discard an arbitrary volume of acknowledged work with nothing visible on a dashboard -- and a backlog exceeding DEFAULT_REQUEST_TTL drops requests in bulk. Adds aggregator_commitments_dropped_total{reason}, covering the pre-existing duplicate and rejected drop paths as well, neither of which was instrumented either. Raises the expiry log from Debug to Warn to match the neighbouring rejected-leaf path, and includes effectiveTimeout so the service-assigned deadline is visible alongside the requester's own. --- deploy/grafana/dashboards/aggregator.json | 27 +++++++++++++++++++++++ internal/metrics/metrics.go | 20 +++++++++++++++++ internal/round/batch_processor.go | 7 +++++- internal/round/leaf_add.go | 2 ++ internal/round/leaf_add_test.go | 6 ++--- internal/round/precollector.go | 6 ++++- 6 files changed, 63 insertions(+), 5 deletions(-) diff --git a/deploy/grafana/dashboards/aggregator.json b/deploy/grafana/dashboards/aggregator.json index ea6c0b1..43e718a 100644 --- a/deploy/grafana/dashboards/aggregator.json +++ b/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 + } + } + } } ] } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 873f348..eda4d4f 100644 --- a/internal/metrics/metrics.go +++ b/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", diff --git a/internal/round/batch_processor.go b/internal/round/batch_processor.go index c45f18a..577d147 100644 --- a/internal/round/batch_processor.go +++ b/internal/round/batch_processor.go @@ -46,10 +46,15 @@ func (rm *RoundManager) processMiniBatchForRound(ctx context.Context, round *Rou 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, diff --git a/internal/round/leaf_add.go b/internal/round/leaf_add.go index 8e8d339..54c2ebf 100644 --- a/internal/round/leaf_add.go +++ b/internal/round/leaf_add.go @@ -87,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, @@ -104,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, diff --git a/internal/round/leaf_add_test.go b/internal/round/leaf_add_test.go index cc8633c..058528a 100644 --- a/internal/round/leaf_add_test.go +++ b/internal/round/leaf_add_test.go @@ -28,7 +28,7 @@ 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) @@ -43,7 +43,7 @@ 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 := materializeCommitmentLeaf(testCommitment(t), referenceTime) @@ -73,7 +73,7 @@ func TestServiceAssignedDeadlineStillBindsReferenceTime(t *testing.T) { // 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 := materializeCommitmentLeaf(commitment, testExpiresAt-1) diff --git a/internal/round/precollector.go b/internal/round/precollector.go index 9b1d499..99344aa 100644 --- a/internal/round/precollector.go +++ b/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" @@ -344,10 +345,13 @@ func (cp *childPrecollector) addBatch( 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,