From 759e57908a6af2567c41c033f9cc55af055a29e4 Mon Sep 17 00:00:00 2001 From: sbackend Date: Sun, 30 Aug 2026 23:40:40 +0200 Subject: [PATCH 1/6] feat: add feed metrics --- openapi/Swarm.yaml | 6 ++-- openapi/SwarmCommon.yaml | 4 ++- pkg/feeds/factory/factory.go | 56 +++++++++++++++++++++++++++++++++--- pkg/node/node.go | 3 ++ 4 files changed, 61 insertions(+), 8 deletions(-) diff --git a/openapi/Swarm.yaml b/openapi/Swarm.yaml index c320410b4ee..4ed796371db 100644 --- a/openapi/Swarm.yaml +++ b/openapi/Swarm.yaml @@ -1,7 +1,7 @@ openapi: 3.0.3 info: - version: 8.1.1 + version: 8.1.2 title: Bee API description: "API endpoints for interacting with the Swarm network, supporting file operations, messaging, and node management" @@ -1114,7 +1114,7 @@ paths: schema: $ref: "SwarmCommon.yaml#/components/schemas/FeedType" required: false - description: "Feed indexing scheme (default: sequence)" + description: "Feed indexing scheme. Only sequence is supported; other values are ignored and sequence is always used." - $ref: "SwarmCommon.yaml#/components/parameters/SwarmPinParameter" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmPostageBatchId" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmAct" @@ -1179,7 +1179,7 @@ paths: schema: $ref: "SwarmCommon.yaml#/components/schemas/FeedType" required: false - description: "Feed indexing scheme (default: sequence)" + description: "Feed indexing scheme. Only sequence is supported; other values are ignored and sequence is always used." - $ref: "SwarmCommon.yaml#/components/parameters/SwarmOnlyRootChunkParameter" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmCache" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyStrategyParameter" diff --git a/openapi/SwarmCommon.yaml b/openapi/SwarmCommon.yaml index ffcbbac3b8e..b7e567b58ae 100644 --- a/openapi/SwarmCommon.yaml +++ b/openapi/SwarmCommon.yaml @@ -866,7 +866,9 @@ components: FeedType: type: string - pattern: "^(sequence|epoch)$" + description: Feed indexing scheme. Only sequence is supported by the HTTP API. + enum: + - sequence IsRetrievableResponse: type: object diff --git a/pkg/feeds/factory/factory.go b/pkg/feeds/factory/factory.go index 1d555416407..5b019a80a70 100644 --- a/pkg/feeds/factory/factory.go +++ b/pkg/feeds/factory/factory.go @@ -5,27 +5,75 @@ package factory import ( + "context" + "errors" + "strings" + "time" + "github.com/ethersphere/bee/v2/pkg/feeds" "github.com/ethersphere/bee/v2/pkg/feeds/epochs" "github.com/ethersphere/bee/v2/pkg/feeds/sequence" storage "github.com/ethersphere/bee/v2/pkg/storage" + "github.com/ethersphere/bee/v2/pkg/swarm" ) type factory struct { storage.Getter + metrics metrics } func New(getter storage.Getter) feeds.Factory { - return &factory{getter} + return &factory{ + Getter: getter, + metrics: newMetrics(), + } } func (f *factory) NewLookup(t feeds.Type, feed *feeds.Feed) (feeds.Lookup, error) { + var lookup feeds.Lookup switch t { case feeds.Sequence: - return sequence.NewAsyncFinder(f.Getter, feed), nil + lookup = sequence.NewAsyncFinder(f.Getter, feed) case feeds.Epoch: - return epochs.NewAsyncFinder(f.Getter, feed), nil + lookup = epochs.NewAsyncFinder(f.Getter, feed) + default: + return nil, feeds.ErrFeedTypeNotFound + } + + return f.wrapLookup(t, lookup), nil +} + +func (f *factory) wrapLookup(t feeds.Type, lookup feeds.Lookup) feeds.Lookup { + return &instrumentedLookup{ + lookup: lookup, + typ: strings.ToLower(t.String()), + m: f.metrics, } +} + +type instrumentedLookup struct { + lookup feeds.Lookup + typ string + m metrics +} - return nil, feeds.ErrFeedTypeNotFound +func (l *instrumentedLookup) At(ctx context.Context, at int64, after uint64) (swarm.Chunk, feeds.Index, feeds.Index, error) { + l.m.LookupStarted.WithLabelValues(l.typ).Inc() + start := time.Now() + ch, cur, next, err := l.lookup.At(ctx, at, after) + l.m.LookupDuration.WithLabelValues(l.typ, lookupResult(ch, err)).Observe(time.Since(start).Seconds()) + return ch, cur, next, err +} + +func lookupResult(ch swarm.Chunk, err error) string { + switch { + case err != nil && errors.Is(err, context.Canceled): + return "canceled" + case err != nil: + return "error" + case ch == nil: + return "not_found" + default: + return "found" + } } diff --git a/pkg/node/node.go b/pkg/node/node.go index c05517ad5be..acef3c9fdca 100644 --- a/pkg/node/node.go +++ b/pkg/node/node.go @@ -1414,6 +1414,9 @@ func NewBee( apiService.MustRegisterMetrics(retrieval.Metrics()...) apiService.MustRegisterMetrics(lightNodes.Metrics()...) apiService.MustRegisterMetrics(hive.Metrics()...) + if feedFactoryMetrics, ok := feedFactory.(metrics.Collector); ok { + apiService.MustRegisterMetrics(feedFactoryMetrics.Metrics()...) + } if bs, ok := batchStore.(metrics.Collector); ok { apiService.MustRegisterMetrics(bs.Metrics()...) From 6ea61ea8c50fa388cbdb73921164ae79e444069d Mon Sep 17 00:00:00 2001 From: sbackend Date: Sun, 30 Aug 2026 23:46:25 +0200 Subject: [PATCH 2/6] fix: missing metrics file --- pkg/feeds/factory/metrics.go | 49 ++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 pkg/feeds/factory/metrics.go diff --git a/pkg/feeds/factory/metrics.go b/pkg/feeds/factory/metrics.go new file mode 100644 index 00000000000..04561d90b5e --- /dev/null +++ b/pkg/feeds/factory/metrics.go @@ -0,0 +1,49 @@ +// Copyright 2026 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package factory + +import ( + "github.com/prometheus/client_golang/prometheus" + + m "github.com/ethersphere/bee/v2/pkg/metrics" +) + +type metrics struct { + // all metrics fields must be exported + // to be able to return them by Metrics() + // using reflection + LookupStarted *prometheus.CounterVec + LookupDuration *prometheus.HistogramVec +} + +func newMetrics() metrics { + subsystem := "feeds" + + return metrics{ + LookupStarted: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: m.Namespace, + Subsystem: subsystem, + Name: "lookup_started_total", + Help: "Number of feed lookup attempts started.", + }, + []string{"type"}, + ), + LookupDuration: prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: m.Namespace, + Subsystem: subsystem, + Name: "lookup_duration_seconds", + Help: "Histogram of feed lookup durations.", + Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, + }, + []string{"type", "result"}, + ), + } +} + +func (f *factory) Metrics() []prometheus.Collector { + return m.PrometheusCollectorsFromFields(f.metrics) +} From 4bc6444ce09db045afa6b552e6c4b6284f9e1ac2 Mon Sep 17 00:00:00 2001 From: sbackend Date: Thu, 10 Sep 2026 16:34:10 +0200 Subject: [PATCH 3/6] fix: clean up PR --- openapi/Swarm.yaml | 6 +++--- openapi/SwarmCommon.yaml | 4 +--- pkg/feeds/factory/factory.go | 17 +++++++++++++++-- pkg/feeds/factory/metrics.go | 2 +- pkg/node/node.go | 4 +--- 5 files changed, 21 insertions(+), 12 deletions(-) diff --git a/openapi/Swarm.yaml b/openapi/Swarm.yaml index 811bc28543c..9140a8b4783 100644 --- a/openapi/Swarm.yaml +++ b/openapi/Swarm.yaml @@ -1,7 +1,7 @@ openapi: 3.0.3 info: - version: 8.2.1 + version: 8.2.0 title: Bee API description: "API endpoints for interacting with the Swarm network, supporting file operations, messaging, and node management" @@ -1114,7 +1114,7 @@ paths: schema: $ref: "SwarmCommon.yaml#/components/schemas/FeedType" required: false - description: "Feed indexing scheme. Only sequence is supported; other values are ignored and sequence is always used." + description: "Feed indexing scheme (default: sequence)" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmPinParameter" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmPostageBatchId" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmAct" @@ -1179,7 +1179,7 @@ paths: schema: $ref: "SwarmCommon.yaml#/components/schemas/FeedType" required: false - description: "Feed indexing scheme. Only sequence is supported; other values are ignored and sequence is always used." + description: "Feed indexing scheme (default: sequence)" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmOnlyRootChunkParameter" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmCache" - $ref: "SwarmCommon.yaml#/components/parameters/SwarmRedundancyStrategyParameter" diff --git a/openapi/SwarmCommon.yaml b/openapi/SwarmCommon.yaml index fb907eb9e66..cdacafb3f7f 100644 --- a/openapi/SwarmCommon.yaml +++ b/openapi/SwarmCommon.yaml @@ -878,9 +878,7 @@ components: FeedType: type: string - description: Feed indexing scheme. Only sequence is supported by the HTTP API. - enum: - - sequence + pattern: "^(sequence|epoch)$" IsRetrievableResponse: type: object diff --git a/pkg/feeds/factory/factory.go b/pkg/feeds/factory/factory.go index 5b019a80a70..44b7e911a2a 100644 --- a/pkg/feeds/factory/factory.go +++ b/pkg/feeds/factory/factory.go @@ -7,16 +7,18 @@ package factory import ( "context" "errors" - "strings" "time" "github.com/ethersphere/bee/v2/pkg/feeds" "github.com/ethersphere/bee/v2/pkg/feeds/epochs" "github.com/ethersphere/bee/v2/pkg/feeds/sequence" + m "github.com/ethersphere/bee/v2/pkg/metrics" storage "github.com/ethersphere/bee/v2/pkg/storage" "github.com/ethersphere/bee/v2/pkg/swarm" ) +var _ m.Collector = (*factory)(nil) + type factory struct { storage.Getter metrics metrics @@ -46,7 +48,7 @@ func (f *factory) NewLookup(t feeds.Type, feed *feeds.Feed) (feeds.Lookup, error func (f *factory) wrapLookup(t feeds.Type, lookup feeds.Lookup) feeds.Lookup { return &instrumentedLookup{ lookup: lookup, - typ: strings.ToLower(t.String()), + typ: lookupType(t), m: f.metrics, } } @@ -65,6 +67,17 @@ func (l *instrumentedLookup) At(ctx context.Context, at int64, after uint64) (sw return ch, cur, next, err } +func lookupType(t feeds.Type) string { + switch t { + case feeds.Sequence: + return "sequence" + case feeds.Epoch: + return "epoch" + default: + return "unknown" + } +} + func lookupResult(ch swarm.Chunk, err error) string { switch { case err != nil && errors.Is(err, context.Canceled): diff --git a/pkg/feeds/factory/metrics.go b/pkg/feeds/factory/metrics.go index 04561d90b5e..95888a1414a 100644 --- a/pkg/feeds/factory/metrics.go +++ b/pkg/feeds/factory/metrics.go @@ -37,7 +37,7 @@ func newMetrics() metrics { Subsystem: subsystem, Name: "lookup_duration_seconds", Help: "Histogram of feed lookup durations.", - Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, + Buckets: []float64{0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 30, 60}, }, []string{"type", "result"}, ), diff --git a/pkg/node/node.go b/pkg/node/node.go index 04f3c5058c9..3c4a45f6b3a 100644 --- a/pkg/node/node.go +++ b/pkg/node/node.go @@ -1414,9 +1414,7 @@ func NewBee( apiService.MustRegisterMetrics(retrieval.Metrics()...) apiService.MustRegisterMetrics(lightNodes.Metrics()...) apiService.MustRegisterMetrics(hive.Metrics()...) - if feedFactoryMetrics, ok := feedFactory.(metrics.Collector); ok { - apiService.MustRegisterMetrics(feedFactoryMetrics.Metrics()...) - } + apiService.MustRegisterMetrics(feedFactory.(metrics.Collector).Metrics()...) if bs, ok := batchStore.(metrics.Collector); ok { apiService.MustRegisterMetrics(bs.Metrics()...) From 2859a6e84cc674d203b6e7137a7bf5be5872f3c3 Mon Sep 17 00:00:00 2001 From: sbackend Date: Sun, 1 Nov 2026 22:55:44 +0100 Subject: [PATCH 4/6] test(feeds): cover instrumented lookup metrics Assert lookupResult labels and that the wrapper records them without changing the lookup result. --- pkg/feeds/factory/export_test.go | 20 ++ pkg/feeds/factory/factory_test.go | 306 ++++++++++++++++++++++++++++++ 2 files changed, 326 insertions(+) create mode 100644 pkg/feeds/factory/export_test.go create mode 100644 pkg/feeds/factory/factory_test.go diff --git a/pkg/feeds/factory/export_test.go b/pkg/feeds/factory/export_test.go new file mode 100644 index 00000000000..c8e818eaa75 --- /dev/null +++ b/pkg/feeds/factory/export_test.go @@ -0,0 +1,20 @@ +// Copyright 2026 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package factory + +import "github.com/ethersphere/bee/v2/pkg/feeds" + +// Factory is an alias for testing. +type Factory = factory + +// WrapLookup exposes wrapLookup for tests. +func (f *factory) WrapLookup(t feeds.Type, lookup feeds.Lookup) feeds.Lookup { + return f.wrapLookup(t, lookup) +} + +// MetricsStruct exposes metrics for tests. +func (f *factory) MetricsStruct() metrics { + return f.metrics +} diff --git a/pkg/feeds/factory/factory_test.go b/pkg/feeds/factory/factory_test.go new file mode 100644 index 00000000000..982c281cc65 --- /dev/null +++ b/pkg/feeds/factory/factory_test.go @@ -0,0 +1,306 @@ +// Copyright 2026 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package factory_test + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethersphere/bee/v2/pkg/feeds" + "github.com/ethersphere/bee/v2/pkg/feeds/factory" + "github.com/ethersphere/bee/v2/pkg/metrics" + "github.com/ethersphere/bee/v2/pkg/swarm" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/expfmt" +) + +func TestFactoryMetricsCollectors(t *testing.T) { + t.Parallel() + + f := factory.New(nil) + c, ok := f.(metrics.Collector) + if !ok { + t.Fatal("factory does not implement metrics.Collector") + } + + collectors := c.Metrics() + if len(collectors) != 2 { + t.Fatalf("got %d collectors, want 2", len(collectors)) + } + + names := make([]string, 0, len(collectors)) + for _, collector := range collectors { + ch := make(chan *prometheus.Desc, 1) + collector.Describe(ch) + close(ch) + for d := range ch { + names = append(names, d.String()) + } + } + + joined := strings.Join(names, "\n") + for _, want := range []string{ + "feeds_lookup_started_total", + "feeds_lookup_duration_seconds", + } { + if !strings.Contains(joined, want) { + t.Errorf("missing metric %q in %s", want, joined) + } + } +} + +func TestInstrumentedLookup(t *testing.T) { + t.Parallel() + + chunk := swarm.NewChunk(swarm.NewAddress(make([]byte, swarm.HashSize)), []byte("payload")) + tests := []struct { + name string + feedType feeds.Type + chunk swarm.Chunk + err error + wantResult string + }{ + { + name: "sequence found", + feedType: feeds.Sequence, + chunk: chunk, + wantResult: "found", + }, + { + name: "sequence not found", + feedType: feeds.Sequence, + wantResult: "not_found", + }, + { + name: "sequence error", + feedType: feeds.Sequence, + err: errors.New("lookup failed"), + wantResult: "error", + }, + { + name: "sequence canceled", + feedType: feeds.Sequence, + err: context.Canceled, + wantResult: "canceled", + }, + { + name: "epoch found", + feedType: feeds.Epoch, + chunk: chunk, + wantResult: "found", + }, + { + name: "epoch not found", + feedType: feeds.Epoch, + wantResult: "not_found", + }, + { + name: "epoch error", + feedType: feeds.Epoch, + err: errors.New("lookup failed"), + wantResult: "error", + }, + { + name: "epoch canceled", + feedType: feeds.Epoch, + err: context.Canceled, + wantResult: "canceled", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + f := factory.New(nil).(*factory.Factory) + lookup := f.WrapLookup(tc.feedType, &mockLookup{chunk: tc.chunk, err: tc.err}) + + ch, _, _, atErr := lookup.At(context.Background(), 0, 0) + if !errors.Is(atErr, tc.err) { + t.Fatalf("got error %v, want %v", atErr, tc.err) + } + if tc.chunk == nil && ch != nil { + t.Fatal("expected nil chunk") + } + if tc.chunk != nil && !ch.Address().Equal(tc.chunk.Address()) { + t.Fatalf("got chunk %s, want %s", ch.Address(), tc.chunk.Address()) + } + + typ := strings.ToLower(tc.feedType.String()) + if got := counterValue(t, f.MetricsStruct().LookupStarted.WithLabelValues(typ)); got != 1 { + t.Fatalf("lookup_started_total{%s}=%v, want 1", typ, got) + } + if got := histogramCount(t, f.MetricsStruct().LookupDuration.WithLabelValues(typ, tc.wantResult)); got != 1 { + t.Fatalf("lookup_duration_seconds{%s,%s} count=%v, want 1", typ, tc.wantResult, got) + } + }) + } +} + +func TestNewLookupUnknownType(t *testing.T) { + t.Parallel() + + f := factory.New(nil) + _, err := f.NewLookup(feeds.Type(99), feeds.New([]byte("topic"), common.Address{})) + if !errors.Is(err, feeds.ErrFeedTypeNotFound) { + t.Fatalf("got %v, want %v", err, feeds.ErrFeedTypeNotFound) + } +} + +func TestMetricsRegistryScrape(t *testing.T) { + t.Parallel() + + f := factory.New(nil).(*factory.Factory) + registry := prometheus.NewRegistry() + for _, c := range f.Metrics() { + if err := registry.Register(c); err != nil { + t.Fatal(err) + } + } + + chunk := swarm.NewChunk(swarm.NewAddress(make([]byte, swarm.HashSize)), []byte("payload")) + lookups := []struct { + typ feeds.Type + chunk swarm.Chunk + err error + }{ + {feeds.Sequence, chunk, nil}, + {feeds.Sequence, nil, nil}, + {feeds.Sequence, nil, errors.New("boom")}, + {feeds.Sequence, nil, context.Canceled}, + {feeds.Epoch, chunk, nil}, + {feeds.Epoch, nil, nil}, + } + for _, tc := range lookups { + _, _, _, _ = f.WrapLookup(tc.typ, &mockLookup{chunk: tc.chunk, err: tc.err}).At(context.Background(), 0, 0) + } + + gathered, err := registry.Gather() + if err != nil { + t.Fatal(err) + } + + got := make(map[string]map[string]float64) + for _, mf := range gathered { + series := make(map[string]float64) + for _, metric := range mf.GetMetric() { + labels := make([]string, 0, len(metric.GetLabel())) + for _, l := range metric.GetLabel() { + labels = append(labels, l.GetName()+"="+l.GetValue()) + } + sort.Strings(labels) + key := strings.Join(labels, ",") + switch mf.GetType() { + case dto.MetricType_COUNTER: + series[key] = metric.GetCounter().GetValue() + case dto.MetricType_HISTOGRAM: + series[key] = float64(metric.GetHistogram().GetSampleCount()) + } + } + got[mf.GetName()] = series + } + + wantStarted := map[string]float64{ + "type=sequence": 4, + "type=epoch": 2, + } + if err := cmpMaps(got["bee_feeds_lookup_started_total"], wantStarted); err != nil { + t.Fatalf("lookup_started_total: %v", err) + } + + wantDuration := map[string]float64{ + "result=canceled,type=sequence": 1, + "result=error,type=sequence": 1, + "result=found,type=epoch": 1, + "result=found,type=sequence": 1, + "result=not_found,type=epoch": 1, + "result=not_found,type=sequence": 1, + } + if err := cmpMaps(got["bee_feeds_lookup_duration_seconds"], wantDuration); err != nil { + t.Fatalf("lookup_duration_seconds: %v\ngot=%v", err, got["bee_feeds_lookup_duration_seconds"]) + } + + var body strings.Builder + enc := expfmt.NewEncoder(&body, expfmt.NewFormat(expfmt.TypeTextPlain)) + for _, mf := range gathered { + if err := enc.Encode(mf); err != nil { + t.Fatal(err) + } + } + if closer, ok := enc.(expfmt.Closer); ok { + _ = closer.Close() + } + text := body.String() + t.Logf("prometheus scrape:\n%s", text) + for _, want := range []string{ + "# HELP bee_feeds_lookup_started_total", + "# TYPE bee_feeds_lookup_started_total counter", + `bee_feeds_lookup_started_total{type="sequence"} 4`, + `bee_feeds_lookup_started_total{type="epoch"} 2`, + "# HELP bee_feeds_lookup_duration_seconds", + "# TYPE bee_feeds_lookup_duration_seconds histogram", + `bee_feeds_lookup_duration_seconds_count{result="found",type="sequence"} 1`, + `bee_feeds_lookup_duration_seconds_count{result="not_found",type="sequence"} 1`, + `bee_feeds_lookup_duration_seconds_count{result="error",type="sequence"} 1`, + `bee_feeds_lookup_duration_seconds_count{result="canceled",type="sequence"} 1`, + `bee_feeds_lookup_duration_seconds_count{result="found",type="epoch"} 1`, + `bee_feeds_lookup_duration_seconds_count{result="not_found",type="epoch"} 1`, + } { + if !strings.Contains(text, want) { + t.Fatalf("scrape missing %q\n%s", want, text) + } + } +} + +type mockLookup struct { + chunk swarm.Chunk + err error +} + +func (m *mockLookup) At(context.Context, int64, uint64) (swarm.Chunk, feeds.Index, feeds.Index, error) { + return m.chunk, nil, nil, m.err +} + +func counterValue(t *testing.T, c prometheus.Counter) float64 { + t.Helper() + metric := &dto.Metric{} + if err := c.Write(metric); err != nil { + t.Fatal(err) + } + return metric.GetCounter().GetValue() +} + +func histogramCount(t *testing.T, o prometheus.Observer) float64 { + t.Helper() + m, ok := o.(prometheus.Metric) + if !ok { + t.Fatal("observer is not a prometheus.Metric") + } + metric := &dto.Metric{} + if err := m.Write(metric); err != nil { + t.Fatal(err) + } + return float64(metric.GetHistogram().GetSampleCount()) +} + +func cmpMaps(got, want map[string]float64) error { + for k, wantV := range want { + if got[k] != wantV { + return fmt.Errorf("key %q got %v want %v", k, got[k], wantV) + } + } + for k := range got { + if _, ok := want[k]; !ok { + return fmt.Errorf("unexpected key %q", k) + } + } + return nil +} From 7452e10bd16e9105ecd61a2ace2616cea1effd85 Mon Sep 17 00:00:00 2001 From: sbackend Date: Mon, 2 Nov 2026 00:10:48 +0100 Subject: [PATCH 5/6] fix: tests audit --- pkg/feeds/factory/export_test.go | 13 +- pkg/feeds/factory/factory_test.go | 383 +++++++++++++++--------------- 2 files changed, 199 insertions(+), 197 deletions(-) diff --git a/pkg/feeds/factory/export_test.go b/pkg/feeds/factory/export_test.go index c8e818eaa75..ef185721387 100644 --- a/pkg/feeds/factory/export_test.go +++ b/pkg/feeds/factory/export_test.go @@ -6,15 +6,6 @@ package factory import "github.com/ethersphere/bee/v2/pkg/feeds" -// Factory is an alias for testing. -type Factory = factory - -// WrapLookup exposes wrapLookup for tests. -func (f *factory) WrapLookup(t feeds.Type, lookup feeds.Lookup) feeds.Lookup { - return f.wrapLookup(t, lookup) -} - -// MetricsStruct exposes metrics for tests. -func (f *factory) MetricsStruct() metrics { - return f.metrics +func WrapLookup(f feeds.Factory, t feeds.Type, lookup feeds.Lookup) feeds.Lookup { + return f.(*factory).wrapLookup(t, lookup) } diff --git a/pkg/feeds/factory/factory_test.go b/pkg/feeds/factory/factory_test.go index 982c281cc65..31232f1e3b3 100644 --- a/pkg/feeds/factory/factory_test.go +++ b/pkg/feeds/factory/factory_test.go @@ -8,54 +8,26 @@ import ( "context" "errors" "fmt" - "sort" + "maps" + "slices" "strings" "testing" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethersphere/bee/v2/pkg/feeds" "github.com/ethersphere/bee/v2/pkg/feeds/factory" "github.com/ethersphere/bee/v2/pkg/metrics" + "github.com/ethersphere/bee/v2/pkg/storage" "github.com/ethersphere/bee/v2/pkg/swarm" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" - "github.com/prometheus/common/expfmt" ) -func TestFactoryMetricsCollectors(t *testing.T) { - t.Parallel() - - f := factory.New(nil) - c, ok := f.(metrics.Collector) - if !ok { - t.Fatal("factory does not implement metrics.Collector") - } - - collectors := c.Metrics() - if len(collectors) != 2 { - t.Fatalf("got %d collectors, want 2", len(collectors)) - } - - names := make([]string, 0, len(collectors)) - for _, collector := range collectors { - ch := make(chan *prometheus.Desc, 1) - collector.Describe(ch) - close(ch) - for d := range ch { - names = append(names, d.String()) - } - } - - joined := strings.Join(names, "\n") - for _, want := range []string{ - "feeds_lookup_started_total", - "feeds_lookup_duration_seconds", - } { - if !strings.Contains(joined, want) { - t.Errorf("missing metric %q in %s", want, joined) - } - } -} +const ( + startedMetric = "bee_feeds_lookup_started_total" + durationMetric = "bee_feeds_lookup_duration_seconds" +) func TestInstrumentedLookup(t *testing.T) { t.Parallel() @@ -66,53 +38,176 @@ func TestInstrumentedLookup(t *testing.T) { feedType feeds.Type chunk swarm.Chunk err error + wantType string wantResult string }{ { - name: "sequence found", + name: "found", feedType: feeds.Sequence, chunk: chunk, + wantType: "sequence", wantResult: "found", }, { - name: "sequence not found", + name: "not found", feedType: feeds.Sequence, + wantType: "sequence", wantResult: "not_found", }, { - name: "sequence error", + name: "error", feedType: feeds.Sequence, err: errors.New("lookup failed"), + wantType: "sequence", wantResult: "error", }, { - name: "sequence canceled", + name: "canceled", feedType: feeds.Sequence, err: context.Canceled, + wantType: "sequence", + wantResult: "canceled", + }, + { + name: "wrapped canceled", + feedType: feeds.Sequence, + err: fmt.Errorf("get chunk: %w", context.Canceled), + wantType: "sequence", wantResult: "canceled", }, + { + name: "deadline exceeded", + feedType: feeds.Sequence, + err: context.DeadlineExceeded, + wantType: "sequence", + wantResult: "error", + }, { name: "epoch found", feedType: feeds.Epoch, chunk: chunk, + wantType: "epoch", wantResult: "found", }, { name: "epoch not found", feedType: feeds.Epoch, + wantType: "epoch", + wantResult: "not_found", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + f := factory.New(nil) + var cur, next feeds.Index = mockIndex("cur"), mockIndex("next") + lookup := factory.WrapLookup(f, tc.feedType, &mockLookup{chunk: tc.chunk, cur: cur, next: next, err: tc.err}) + + gotChunk, gotCur, gotNext, err := lookup.At(context.Background(), 0, 0) + if !errors.Is(err, tc.err) { + t.Fatalf("got error %v, want %v", err, tc.err) + } + if gotChunk != tc.chunk { + t.Fatalf("got chunk %v, want %v", gotChunk, tc.chunk) + } + if gotCur != cur || gotNext != next { + t.Fatalf("got indexes %v/%v, want %v/%v", gotCur, gotNext, cur, next) + } + + counts, _ := gather(t, f) + want := map[string]float64{ + startedMetric + "{type=" + tc.wantType + "}": 1, + durationMetric + "_count{result=" + tc.wantResult + ",type=" + tc.wantType + "}": 1, + } + if !maps.Equal(counts, want) { + t.Fatalf("got metrics %v, want %v", counts, want) + } + }) + } +} + +func TestInstrumentedLookupDuration(t *testing.T) { + t.Parallel() + + const delay = 50 * time.Millisecond + f := factory.New(nil) + lookup := factory.WrapLookup(f, feeds.Sequence, &mockLookup{delay: delay}) + + if _, _, _, err := lookup.At(context.Background(), 0, 0); err != nil { + t.Fatal(err) + } + + _, sums := gather(t, f) + key := durationMetric + "_sum{result=not_found,type=sequence}" + if got := sums[key]; got < delay.Seconds() { + t.Fatalf("%s=%v, want at least %v", key, got, delay.Seconds()) + } +} + +func TestInstrumentedLookupStartedBeforeCompletion(t *testing.T) { + t.Parallel() + + f := factory.New(nil) + entered := make(chan struct{}) + release := make(chan struct{}) + lookup := factory.WrapLookup(f, feeds.Sequence, &mockLookup{entered: entered, release: release}) + + done := make(chan struct{}) + go func() { + defer close(done) + _, _, _, _ = lookup.At(context.Background(), 0, 0) + }() + + <-entered + inFlight, _ := gather(t, f) + close(release) + <-done + + want := map[string]float64{startedMetric + "{type=sequence}": 1} + if !maps.Equal(inFlight, want) { + t.Fatalf("got metrics during lookup %v, want %v", inFlight, want) + } +} + +func TestNewLookupRecordsMetrics(t *testing.T) { + t.Parallel() + + notFound := storage.GetterFunc(func(context.Context, swarm.Address) (swarm.Chunk, error) { + return nil, storage.ErrNotFound + }) + failing := storage.GetterFunc(func(context.Context, swarm.Address) (swarm.Chunk, error) { + return nil, errors.New("getter failed") + }) + + tests := []struct { + name string + feedType feeds.Type + getter storage.Getter + wantType string + wantResult string + }{ + { + name: "sequence not found", + feedType: feeds.Sequence, + getter: notFound, + wantType: "sequence", wantResult: "not_found", }, { - name: "epoch error", - feedType: feeds.Epoch, - err: errors.New("lookup failed"), + name: "sequence error", + feedType: feeds.Sequence, + getter: failing, + wantType: "sequence", wantResult: "error", }, { - name: "epoch canceled", + name: "epoch not found", feedType: feeds.Epoch, - err: context.Canceled, - wantResult: "canceled", + getter: notFound, + wantType: "epoch", + wantResult: "not_found", }, } @@ -120,26 +215,20 @@ func TestInstrumentedLookup(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() - f := factory.New(nil).(*factory.Factory) - lookup := f.WrapLookup(tc.feedType, &mockLookup{chunk: tc.chunk, err: tc.err}) - - ch, _, _, atErr := lookup.At(context.Background(), 0, 0) - if !errors.Is(atErr, tc.err) { - t.Fatalf("got error %v, want %v", atErr, tc.err) - } - if tc.chunk == nil && ch != nil { - t.Fatal("expected nil chunk") - } - if tc.chunk != nil && !ch.Address().Equal(tc.chunk.Address()) { - t.Fatalf("got chunk %s, want %s", ch.Address(), tc.chunk.Address()) + f := factory.New(tc.getter) + lookup, err := f.NewLookup(tc.feedType, feeds.New([]byte("topic"), common.Address{})) + if err != nil { + t.Fatal(err) } + _, _, _, _ = lookup.At(context.Background(), time.Now().Unix(), 0) - typ := strings.ToLower(tc.feedType.String()) - if got := counterValue(t, f.MetricsStruct().LookupStarted.WithLabelValues(typ)); got != 1 { - t.Fatalf("lookup_started_total{%s}=%v, want 1", typ, got) + counts, _ := gather(t, f) + want := map[string]float64{ + startedMetric + "{type=" + tc.wantType + "}": 1, + durationMetric + "_count{result=" + tc.wantResult + ",type=" + tc.wantType + "}": 1, } - if got := histogramCount(t, f.MetricsStruct().LookupDuration.WithLabelValues(typ, tc.wantResult)); got != 1 { - t.Fatalf("lookup_duration_seconds{%s,%s} count=%v, want 1", typ, tc.wantResult, got) + if !maps.Equal(counts, want) { + t.Fatalf("got metrics %v, want %v", counts, want) } }) } @@ -153,154 +242,76 @@ func TestNewLookupUnknownType(t *testing.T) { if !errors.Is(err, feeds.ErrFeedTypeNotFound) { t.Fatalf("got %v, want %v", err, feeds.ErrFeedTypeNotFound) } + + if counts, _ := gather(t, f); len(counts) != 0 { + t.Fatalf("got metrics %v, want none", counts) + } } -func TestMetricsRegistryScrape(t *testing.T) { - t.Parallel() +// gather registers the factory metrics in a fresh registry and returns +// counter values and histogram sample counts, and histogram sample sums, +// keyed by metric name and sorted labels. +func gather(t *testing.T, f feeds.Factory) (counts, sums map[string]float64) { + t.Helper() - f := factory.New(nil).(*factory.Factory) + c, ok := f.(metrics.Collector) + if !ok { + t.Fatal("factory does not implement metrics.Collector") + } registry := prometheus.NewRegistry() - for _, c := range f.Metrics() { - if err := registry.Register(c); err != nil { + for _, collector := range c.Metrics() { + if err := registry.Register(collector); err != nil { t.Fatal(err) } } - - chunk := swarm.NewChunk(swarm.NewAddress(make([]byte, swarm.HashSize)), []byte("payload")) - lookups := []struct { - typ feeds.Type - chunk swarm.Chunk - err error - }{ - {feeds.Sequence, chunk, nil}, - {feeds.Sequence, nil, nil}, - {feeds.Sequence, nil, errors.New("boom")}, - {feeds.Sequence, nil, context.Canceled}, - {feeds.Epoch, chunk, nil}, - {feeds.Epoch, nil, nil}, - } - for _, tc := range lookups { - _, _, _, _ = f.WrapLookup(tc.typ, &mockLookup{chunk: tc.chunk, err: tc.err}).At(context.Background(), 0, 0) - } - - gathered, err := registry.Gather() + families, err := registry.Gather() if err != nil { t.Fatal(err) } - got := make(map[string]map[string]float64) - for _, mf := range gathered { - series := make(map[string]float64) - for _, metric := range mf.GetMetric() { - labels := make([]string, 0, len(metric.GetLabel())) - for _, l := range metric.GetLabel() { + counts = make(map[string]float64) + sums = make(map[string]float64) + for _, mf := range families { + for _, m := range mf.GetMetric() { + labels := make([]string, 0, len(m.GetLabel())) + for _, l := range m.GetLabel() { labels = append(labels, l.GetName()+"="+l.GetValue()) } - sort.Strings(labels) - key := strings.Join(labels, ",") + slices.Sort(labels) + key := "{" + strings.Join(labels, ",") + "}" + switch mf.GetType() { case dto.MetricType_COUNTER: - series[key] = metric.GetCounter().GetValue() + counts[mf.GetName()+key] = m.GetCounter().GetValue() case dto.MetricType_HISTOGRAM: - series[key] = float64(metric.GetHistogram().GetSampleCount()) + counts[mf.GetName()+"_count"+key] = float64(m.GetHistogram().GetSampleCount()) + sums[mf.GetName()+"_sum"+key] = m.GetHistogram().GetSampleSum() } } - got[mf.GetName()] = series } + return counts, sums +} - wantStarted := map[string]float64{ - "type=sequence": 4, - "type=epoch": 2, - } - if err := cmpMaps(got["bee_feeds_lookup_started_total"], wantStarted); err != nil { - t.Fatalf("lookup_started_total: %v", err) - } +type mockIndex string - wantDuration := map[string]float64{ - "result=canceled,type=sequence": 1, - "result=error,type=sequence": 1, - "result=found,type=epoch": 1, - "result=found,type=sequence": 1, - "result=not_found,type=epoch": 1, - "result=not_found,type=sequence": 1, - } - if err := cmpMaps(got["bee_feeds_lookup_duration_seconds"], wantDuration); err != nil { - t.Fatalf("lookup_duration_seconds: %v\ngot=%v", err, got["bee_feeds_lookup_duration_seconds"]) - } - - var body strings.Builder - enc := expfmt.NewEncoder(&body, expfmt.NewFormat(expfmt.TypeTextPlain)) - for _, mf := range gathered { - if err := enc.Encode(mf); err != nil { - t.Fatal(err) - } - } - if closer, ok := enc.(expfmt.Closer); ok { - _ = closer.Close() - } - text := body.String() - t.Logf("prometheus scrape:\n%s", text) - for _, want := range []string{ - "# HELP bee_feeds_lookup_started_total", - "# TYPE bee_feeds_lookup_started_total counter", - `bee_feeds_lookup_started_total{type="sequence"} 4`, - `bee_feeds_lookup_started_total{type="epoch"} 2`, - "# HELP bee_feeds_lookup_duration_seconds", - "# TYPE bee_feeds_lookup_duration_seconds histogram", - `bee_feeds_lookup_duration_seconds_count{result="found",type="sequence"} 1`, - `bee_feeds_lookup_duration_seconds_count{result="not_found",type="sequence"} 1`, - `bee_feeds_lookup_duration_seconds_count{result="error",type="sequence"} 1`, - `bee_feeds_lookup_duration_seconds_count{result="canceled",type="sequence"} 1`, - `bee_feeds_lookup_duration_seconds_count{result="found",type="epoch"} 1`, - `bee_feeds_lookup_duration_seconds_count{result="not_found",type="epoch"} 1`, - } { - if !strings.Contains(text, want) { - t.Fatalf("scrape missing %q\n%s", want, text) - } - } -} +func (i mockIndex) MarshalBinary() ([]byte, error) { return []byte(i), nil } +func (i mockIndex) Next(int64, uint64) feeds.Index { return i } +func (i mockIndex) String() string { return string(i) } type mockLookup struct { - chunk swarm.Chunk - err error + chunk swarm.Chunk + cur, next feeds.Index + err error + delay time.Duration + entered chan struct{} + release chan struct{} } func (m *mockLookup) At(context.Context, int64, uint64) (swarm.Chunk, feeds.Index, feeds.Index, error) { - return m.chunk, nil, nil, m.err -} - -func counterValue(t *testing.T, c prometheus.Counter) float64 { - t.Helper() - metric := &dto.Metric{} - if err := c.Write(metric); err != nil { - t.Fatal(err) - } - return metric.GetCounter().GetValue() -} - -func histogramCount(t *testing.T, o prometheus.Observer) float64 { - t.Helper() - m, ok := o.(prometheus.Metric) - if !ok { - t.Fatal("observer is not a prometheus.Metric") - } - metric := &dto.Metric{} - if err := m.Write(metric); err != nil { - t.Fatal(err) - } - return float64(metric.GetHistogram().GetSampleCount()) -} - -func cmpMaps(got, want map[string]float64) error { - for k, wantV := range want { - if got[k] != wantV { - return fmt.Errorf("key %q got %v want %v", k, got[k], wantV) - } - } - for k := range got { - if _, ok := want[k]; !ok { - return fmt.Errorf("unexpected key %q", k) - } + if m.entered != nil { + close(m.entered) + <-m.release } - return nil + time.Sleep(m.delay) + return m.chunk, m.cur, m.next, m.err } From 85c88db80e9baac4317ae9a3505b8cb6002377d5 Mon Sep 17 00:00:00 2001 From: sbackend Date: Wed, 4 Nov 2026 23:48:00 +0100 Subject: [PATCH 6/6] fix: address pr comments --- pkg/feeds/factory/factory.go | 8 ++--- pkg/feeds/factory/factory_test.go | 57 ++++++++++++++++++++++++++++++- 2 files changed, 60 insertions(+), 5 deletions(-) diff --git a/pkg/feeds/factory/factory.go b/pkg/feeds/factory/factory.go index 44b7e911a2a..ce48b7f57ca 100644 --- a/pkg/feeds/factory/factory.go +++ b/pkg/feeds/factory/factory.go @@ -63,7 +63,7 @@ func (l *instrumentedLookup) At(ctx context.Context, at int64, after uint64) (sw l.m.LookupStarted.WithLabelValues(l.typ).Inc() start := time.Now() ch, cur, next, err := l.lookup.At(ctx, at, after) - l.m.LookupDuration.WithLabelValues(l.typ, lookupResult(ch, err)).Observe(time.Since(start).Seconds()) + l.m.LookupDuration.WithLabelValues(l.typ, lookupResult(ctx, ch, err)).Observe(time.Since(start).Seconds()) return ch, cur, next, err } @@ -78,11 +78,11 @@ func lookupType(t feeds.Type) string { } } -func lookupResult(ch swarm.Chunk, err error) string { +func lookupResult(ctx context.Context, ch swarm.Chunk, err error) string { switch { - case err != nil && errors.Is(err, context.Canceled): + case errors.Is(ctx.Err(), context.Canceled) || errors.Is(err, context.Canceled): return "canceled" - case err != nil: + case err != nil || ctx.Err() != nil: return "error" case ch == nil: return "not_found" diff --git a/pkg/feeds/factory/factory_test.go b/pkg/feeds/factory/factory_test.go index 31232f1e3b3..9876f8cc7e3 100644 --- a/pkg/feeds/factory/factory_test.go +++ b/pkg/feeds/factory/factory_test.go @@ -33,11 +33,14 @@ func TestInstrumentedLookup(t *testing.T) { t.Parallel() chunk := swarm.NewChunk(swarm.NewAddress(make([]byte, swarm.HashSize)), []byte("payload")) + expired, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + t.Cleanup(cancel) tests := []struct { name string feedType feeds.Type chunk swarm.Chunk err error + ctx context.Context wantType string wantResult string }{ @@ -82,6 +85,14 @@ func TestInstrumentedLookup(t *testing.T) { wantType: "sequence", wantResult: "error", }, + { + name: "deadline context", + feedType: feeds.Sequence, + chunk: chunk, + ctx: expired, + wantType: "sequence", + wantResult: "error", + }, { name: "epoch found", feedType: feeds.Epoch, @@ -105,7 +116,11 @@ func TestInstrumentedLookup(t *testing.T) { var cur, next feeds.Index = mockIndex("cur"), mockIndex("next") lookup := factory.WrapLookup(f, tc.feedType, &mockLookup{chunk: tc.chunk, cur: cur, next: next, err: tc.err}) - gotChunk, gotCur, gotNext, err := lookup.At(context.Background(), 0, 0) + ctx := tc.ctx + if ctx == nil { + ctx = context.Background() + } + gotChunk, gotCur, gotNext, err := lookup.At(ctx, 0, 0) if !errors.Is(err, tc.err) { t.Fatalf("got error %v, want %v", err, tc.err) } @@ -234,6 +249,46 @@ func TestNewLookupRecordsMetrics(t *testing.T) { } } +func TestNewLookupSequenceCanceledAfterFirstChunk(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + chunk := swarm.NewChunk(swarm.NewAddress(make([]byte, swarm.HashSize)), []byte("payload")) + getter := storage.GetterFunc(func(ctx context.Context, _ swarm.Address) (swarm.Chunk, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + // The client goes away after the first update is fetched. Later probes + // observe the canceled context and the sequence finder treats them as + // not found, returning this chunk with a nil error. + cancel() + return chunk, nil + }) + + f := factory.New(getter) + lookup, err := f.NewLookup(feeds.Sequence, feeds.New([]byte("topic"), common.Address{})) + if err != nil { + t.Fatal(err) + } + + got, _, _, err := lookup.At(ctx, time.Now().Unix(), 0) + if err != nil { + t.Fatalf("got error %v, want nil", err) + } + if got != chunk { + t.Fatalf("got chunk %v, want the chunk fetched before cancel", got) + } + + counts, _ := gather(t, f) + want := map[string]float64{ + startedMetric + "{type=sequence}": 1, + durationMetric + "_count{result=canceled,type=sequence}": 1, + } + if !maps.Equal(counts, want) { + t.Fatalf("got metrics %v, want %v", counts, want) + } +} + func TestNewLookupUnknownType(t *testing.T) { t.Parallel()