From e338fbcc9ec11185a9f0a8af0c2c0abda0738bcb Mon Sep 17 00:00:00 2001 From: jait91 Date: Fri, 19 Jun 2026 13:50:22 +0300 Subject: [PATCH 1/3] fix: persisted-nodes gauge --- internal/metrics/metrics.go | 2 +- internal/round/round_manager.go | 6 ++++-- internal/storage/interfaces/interfaces.go | 3 +++ internal/storage/mongodb/aggregator_record.go | 9 +++++++++ 4 files changed, 17 insertions(+), 3 deletions(-) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 90f162c..873f348 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -347,7 +347,7 @@ var ( SMTNodesPersistedTotal = promauto.NewGaugeFunc( prometheus.GaugeOpts{ Name: "aggregator_smt_nodes_persisted_total", - Help: "Approximate persisted SMT node count from MongoDB EstimatedDocumentCount.", + Help: "Approximate finalized aggregator-record (SMT leaf) count per shard via Mongo EstimatedDocumentCount; a per-shard growth proxy (leaves, not RocksDB internal nodes).", }, getSMTNodesPersistedCount, ) diff --git a/internal/round/round_manager.go b/internal/round/round_manager.go index d8198ed..3f33d18 100644 --- a/internal/round/round_manager.go +++ b/internal/round/round_manager.go @@ -233,9 +233,11 @@ func NewRoundManagerWithBackend( avgSMTUpdateTime: 5 * time.Millisecond, // Initial estimate per batch } - if rm.storage != nil && rm.storage.SmtStorage() != nil { + // Count records (SMT leaves) in Mongo; disk-backed SMT nodes live in RocksDB, + // so the SMT collection reads ~0. + if rm.storage != nil && rm.storage.AggregatorRecordStorage() != nil { metrics.SetSMTNodesPersistedCountFunc(func(countCtx context.Context) (int64, error) { - return rm.storage.SmtStorage().EstimatedCount(countCtx) + return rm.storage.AggregatorRecordStorage().EstimatedCount(countCtx) }) } if rm.commitmentQueue != nil { diff --git a/internal/storage/interfaces/interfaces.go b/internal/storage/interfaces/interfaces.go index 0889d4c..d80baff 100644 --- a/internal/storage/interfaces/interfaces.go +++ b/internal/storage/interfaces/interfaces.go @@ -78,6 +78,9 @@ type AggregatorRecordStorage interface { // Count returns the total number of records Count(ctx context.Context) (int64, error) + // EstimatedCount returns an approximate record count from collection metadata (O(1)). + EstimatedCount(ctx context.Context) (int64, error) + // GetExistingStateIDs returns which of the given state IDs already exist. GetExistingStateIDs(ctx context.Context, stateIDs []string) (map[string]bool, error) } diff --git a/internal/storage/mongodb/aggregator_record.go b/internal/storage/mongodb/aggregator_record.go index 775be11..e82070b 100644 --- a/internal/storage/mongodb/aggregator_record.go +++ b/internal/storage/mongodb/aggregator_record.go @@ -367,6 +367,15 @@ func (ars *AggregatorRecordStorage) Count(ctx context.Context) (int64, error) { return count, nil } +// EstimatedCount returns an approximate record count from collection metadata (O(1), unlike Count's scan). +func (ars *AggregatorRecordStorage) EstimatedCount(ctx context.Context) (int64, error) { + count, err := ars.collection.EstimatedDocumentCount(ctx) + if err != nil { + return 0, fmt.Errorf("failed to estimate aggregator record count: %w", err) + } + return count, nil +} + // CreateIndexes creates the necessary indexes needed by the submit, proof, and // block-record lookup paths. func (ars *AggregatorRecordStorage) CreateIndexes(ctx context.Context) error { From 3618906d69a66a41b692981497f20ed2214f49d3 Mon Sep 17 00:00:00 2001 From: jait91 Date: Fri, 19 Jun 2026 14:00:23 +0300 Subject: [PATCH 2/3] fix tests --- internal/ha/block_syncer_test.go | 4 ++++ internal/service/service_test.go | 3 ++- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/internal/ha/block_syncer_test.go b/internal/ha/block_syncer_test.go index 83b71ee..6140b36 100644 --- a/internal/ha/block_syncer_test.go +++ b/internal/ha/block_syncer_test.go @@ -557,6 +557,10 @@ func (s *blockSyncerTestAggregatorRecordStorage) Count(context.Context) (int64, return count, nil } +func (s *blockSyncerTestAggregatorRecordStorage) EstimatedCount(ctx context.Context) (int64, error) { + return s.Count(ctx) +} + func (s *blockSyncerTestAggregatorRecordStorage) GetExistingStateIDs(_ context.Context, stateIDs []string) (map[string]bool, error) { out := make(map[string]bool, len(stateIDs)) wanted := make(map[string]struct{}, len(stateIDs)) diff --git a/internal/service/service_test.go b/internal/service/service_test.go index 7aaf5e6..7eb3026 100644 --- a/internal/service/service_test.go +++ b/internal/service/service_test.go @@ -1210,7 +1210,8 @@ func (s *testAggregatorRecordStorage) StoreBatch(context.Context, []*models.Aggr func (s *testAggregatorRecordStorage) GetByBlockNumber(context.Context, *api.BigInt) ([]*models.AggregatorRecord, error) { return nil, nil } -func (s *testAggregatorRecordStorage) Count(context.Context) (int64, error) { return 0, nil } +func (s *testAggregatorRecordStorage) Count(context.Context) (int64, error) { return 0, nil } +func (s *testAggregatorRecordStorage) EstimatedCount(context.Context) (int64, error) { return 0, nil } func (s *testAggregatorRecordStorage) GetExistingStateIDs(context.Context, []string) (map[string]bool, error) { return nil, nil } From 9406d63aaf194dd74e19dd9c2741aeb278b70960 Mon Sep 17 00:00:00 2001 From: jait91 Date: Fri, 19 Jun 2026 15:47:28 +0300 Subject: [PATCH 3/3] fix test --- internal/round/disk_smt_startup_test.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/internal/round/disk_smt_startup_test.go b/internal/round/disk_smt_startup_test.go index 548a631..ac26826 100644 --- a/internal/round/disk_smt_startup_test.go +++ b/internal/round/disk_smt_startup_test.go @@ -701,6 +701,10 @@ func (s *diskStartupAggregatorRecordStorage) Count(context.Context) (int64, erro return int64(len(s.records)), nil } +func (s *diskStartupAggregatorRecordStorage) EstimatedCount(context.Context) (int64, error) { + return int64(len(s.records)), nil +} + func (s *diskStartupAggregatorRecordStorage) GetExistingStateIDs(_ context.Context, stateIDs []string) (map[string]bool, error) { existing := make(map[string]bool, len(stateIDs)) for _, stateID := range stateIDs {