diff --git a/deploy/grafana/dashboards/aggregator.json b/deploy/grafana/dashboards/aggregator.json index ea6c0b1c..43e718a4 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/config/config.go b/internal/config/config.go index 017eaf5d..173e4212 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"), diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 873f3483..eda4d4f0 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 c3ebcda0..577d147a 100644 --- a/internal/round/batch_processor.go +++ b/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, diff --git a/internal/round/disk_bft_integration_rocksdb_test.go b/internal/round/disk_bft_integration_rocksdb_test.go index d57c402c..a957fbf0 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 a51dd47f..88a3ab0c 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 3b4a90b3..54c2ebfc 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 } @@ -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, diff --git a/internal/round/leaf_add_test.go b/internal/round/leaf_add_test.go index 398fd524..058528ac 100644 --- a/internal/round/leaf_add_test.go +++ b/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) } diff --git a/internal/round/precollector.go b/internal/round/precollector.go index c2922c27..99344aa1 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" @@ -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, diff --git a/internal/round/round_process_regression_test.go b/internal/round/round_process_regression_test.go index 258d3a45..3987d06a 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}) diff --git a/pkg/api/certification_data_decode_test.go b/pkg/api/certification_data_decode_test.go new file mode 100644 index 00000000..9c7cc14e --- /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 0d86dd55..56562b80 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