From bc4627ebbaebb175effdcbcf7db63e1a5e07e175 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 14:35:51 +0200 Subject: [PATCH 01/11] fix(storer): stop direct upload at the first chunk error --- pkg/storer/netstore.go | 11 ++++++- pkg/storer/netstore_test.go | 59 +++++++++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 1 deletion(-) diff --git a/pkg/storer/netstore.go b/pkg/storer/netstore.go index bf7d3fc5541..5be414a8395 100644 --- a/pkg/storer/netstore.go +++ b/pkg/storer/netstore.go @@ -33,7 +33,16 @@ func (db *DB) DirectUpload() PutterSession { return &putterSession{ Putter: putterWithMetrics{ storage.PutterFunc(func(ctx context.Context, ch swarm.Chunk) error { - db.directUploadLimiter <- struct{}{} + // the upload has already failed: stop it here instead of + // pushing the rest of its chunks + if egCtx.Err() != nil { + return context.Cause(egCtx) + } + select { + case db.directUploadLimiter <- struct{}{}: + case <-egCtx.Done(): + return context.Cause(egCtx) + } eg.Go(safe.RunFunc(db.logger, "storer-netstore-direct-upload", func() (err error) { defer func() { <-db.directUploadLimiter }() diff --git a/pkg/storer/netstore_test.go b/pkg/storer/netstore_test.go index 302d0c055f6..59940c8ce2e 100644 --- a/pkg/storer/netstore_test.go +++ b/pkg/storer/netstore_test.go @@ -10,6 +10,7 @@ import ( "fmt" "slices" "testing" + "testing/synctest" "time" "github.com/ethersphere/bee/v2/pkg/pusher" @@ -388,3 +389,61 @@ func TestNetStore(t *testing.T) { }) }) } + +// TestDirectUploadStopsOnError verifies that once a chunk fails a direct +// upload, the upload stops: later chunks are not pushed and the next put +// returns the error. +func TestDirectUploadStopsOnError(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + chunks := chunktesting.GenerateTestRandomChunks(2) + + lstore, err := storer.New(context.Background(), "", dbTestOps(swarm.RandAddress(t), 0, nil, nil, time.Second)) + if err != nil { + t.Fatal(err) + } + + var pushed []swarm.Address + quit := make(chan struct{}) + defer func() { + close(quit) + if err := lstore.Close(); err != nil { + t.Errorf("close: %v", err) + } + // leveldb waits up to a second for its memory pool on close + time.Sleep(time.Second) + }() + go func() { + for { + select { + case op := <-lstore.PusherFeed(): + pushed = append(pushed, op.Chunk.Address()) + op.Err <- pushsync.ErrNoPush + case <-quit: + return + } + } + }() + + session := lstore.DirectUpload() + + if err := session.Put(context.Background(), chunks[0]); err != nil { + t.Fatalf("session.Put(...): unexpected error: %v", err) + } + synctest.Wait() + + if err := session.Put(context.Background(), chunks[1]); !errors.Is(err, pushsync.ErrNoPush) { + t.Fatalf("session.Put(...): got %v, want %v", err, pushsync.ErrNoPush) + } + synctest.Wait() + + if len(pushed) != 1 || !pushed[0].Equal(chunks[0].Address()) { + t.Fatalf("pushed %v, want only %s", pushed, chunks[0].Address()) + } + + if err := session.Done(chunks[1].Address()); !errors.Is(err, pushsync.ErrNoPush) { + t.Fatalf("session.Done(): got %v, want %v", err, pushsync.ErrNoPush) + } + }) +} From d96a5a29ac7fbec749bf6e1ba55f9f357a4cf496 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 14:36:25 +0200 Subject: [PATCH 02/11] fix(pusher): reset attempts when a direct chunk is out of depth --- pkg/pusher/pusher.go | 1 + pkg/pusher/pusher_test.go | 44 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+) diff --git a/pkg/pusher/pusher.go b/pkg/pusher/pusher.go index 7b1b04389d1..a3afa62389f 100644 --- a/pkg/pusher/pusher.go +++ b/pkg/pusher/pusher.go @@ -370,6 +370,7 @@ func (s *Service) pushDirect(ctx context.Context, logger log.Logger, op *Op) err s.metrics.TotalCouldNotSync.Inc() loggerV1.Debug("direct upload: no node within the chunk's neighborhood took it", "chunk_address", op.Chunk.Address()) err = ErrCouldNotSync + s.attempts.delete(op.identityAddress) case err == nil: s.metrics.TotalSynced.Inc() s.attempts.delete(op.identityAddress) diff --git a/pkg/pusher/pusher_test.go b/pkg/pusher/pusher_test.go index bbf0888c16c..ee64df369a3 100644 --- a/pkg/pusher/pusher_test.go +++ b/pkg/pusher/pusher_test.go @@ -863,3 +863,47 @@ func createPusher( return pusherService } + +// TestDirectOutOfDepthResetsAttempts tests that a direct chunk which ends with +// ErrOutOfDepthStoring gets the full retry budget when it is pushed again. +func TestDirectOutOfDepthResetsAttempts(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { + if calls.Add(1) == 2 { + return nil, pushsync.ErrOutOfDepthStoring + } + return &pushsync.Receipt{Address: chunk.Address()}, pushsync.ErrShallowReceipt + }) + + storer := &mockStorer{chunks: make(chan swarm.Chunk)} + pusherSvc := createPusher(t, storer, pushSyncService, defaultMockBatchStore, defaultRetryCount) + + newFeed := make(chan *pusher.Op) + pusherSvc.AddFeed(newFeed) + + chunk := testingc.GenerateTestRandomChunk() + push := func() error { + errC := make(chan error, 1) + newFeed <- &pusher.Op{Chunk: chunk, Err: errC, Direct: true} + return <-errC + } + + if err := push(); !errors.Is(err, pushsync.ErrShallowReceipt) { + t.Fatalf("push 1: got %v, want %v", err, pushsync.ErrShallowReceipt) + } + if err := push(); !errors.Is(err, pusher.ErrCouldNotSync) { + t.Fatalf("push 2: got %v, want %v", err, pusher.ErrCouldNotSync) + } + + // the next round has the full budget again + for i := 1; i < defaultRetryCount; i++ { + if err := push(); !errors.Is(err, pushsync.ErrShallowReceipt) { + t.Fatalf("round 2, push %d: got %v, want %v", i, err, pushsync.ErrShallowReceipt) + } + } + if err := push(); !errors.Is(err, pusher.ErrCouldNotSync) { + t.Fatalf("round 2, last push: got %v, want %v", err, pusher.ErrCouldNotSync) + } +} From c6976c0ccefcda5487b1c1e76221b04d660f2cad Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 14:37:05 +0200 Subject: [PATCH 03/11] fix(pusher): fail a direct push whose batch no longer exists --- pkg/pusher/pusher.go | 5 +++++ pkg/pusher/pusher_test.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/pkg/pusher/pusher.go b/pkg/pusher/pusher.go index a3afa62389f..a879c45bc2c 100644 --- a/pkg/pusher/pusher.go +++ b/pkg/pusher/pusher.go @@ -12,6 +12,7 @@ import ( "context" "encoding/hex" "errors" + "fmt" "sync" "time" @@ -341,6 +342,10 @@ func (s *Service) pushDirect(ctx context.Context, logger log.Logger, op *Op) err "chunk_address", op.Chunk.Address(), "error", err, ) + if err == nil { + // the chunk was not sent, so it must not count as delivered + err = fmt.Errorf("pusher: batch %s: %w", hex.EncodeToString(op.Chunk.Stamp().BatchID()), postage.ErrNotFound) + } return err } diff --git a/pkg/pusher/pusher_test.go b/pkg/pusher/pusher_test.go index ee64df369a3..755d16dea8b 100644 --- a/pkg/pusher/pusher_test.go +++ b/pkg/pusher/pusher_test.go @@ -907,3 +907,34 @@ func TestDirectOutOfDepthResetsAttempts(t *testing.T) { t.Fatalf("round 2, last push: got %v, want %v", err, pusher.ErrCouldNotSync) } } + +// TestDirectBatchGone tests that a direct chunk whose batch no longer exists +// fails instead of counting as delivered. +func TestDirectBatchGone(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { + calls.Add(1) + return &pushsync.Receipt{Address: chunk.Address()}, nil + }) + bmock := batchstoremock.New(batchstoremock.WithExistsFunc(func(b []byte) (bool, error) { + return false, nil + })) + + storer := &mockStorer{chunks: make(chan swarm.Chunk)} + pusherSvc := createPusher(t, storer, pushSyncService, bmock, defaultRetryCount) + + newFeed := make(chan *pusher.Op) + pusherSvc.AddFeed(newFeed) + + errC := make(chan error, 1) + newFeed <- &pusher.Op{Chunk: testingc.GenerateTestRandomChunk(), Err: errC, Direct: true} + + if err := <-errC; !errors.Is(err, postage.ErrNotFound) { + t.Fatalf("got %v, want %v", err, postage.ErrNotFound) + } + if got := calls.Load(); got != 0 { + t.Fatalf("got %d pushes, want 0", got) + } +} From 2885388d05d8e976e21f45cb557da4bc8b2afa3e Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 14:39:30 +0200 Subject: [PATCH 04/11] fix(storer): retry a failed direct upload chunk at the end of the upload --- pkg/storer/export_test.go | 23 ++++ pkg/storer/metrics.go | 18 +++ pkg/storer/netstore.go | 154 +++++++++++++++------- pkg/storer/netstore_test.go | 252 ++++++++++++++++++++++++++++++++++++ pkg/storer/requeue.go | 60 +++++++++ pkg/storer/requeue_test.go | 88 +++++++++++++ 6 files changed, 548 insertions(+), 47 deletions(-) create mode 100644 pkg/storer/requeue.go create mode 100644 pkg/storer/requeue_test.go diff --git a/pkg/storer/export_test.go b/pkg/storer/export_test.go index d9b779ba97f..08c03aa3b1e 100644 --- a/pkg/storer/export_test.go +++ b/pkg/storer/export_test.go @@ -5,6 +5,8 @@ package storer import ( + "context" + "github.com/ethersphere/bee/v2/pkg/bmt" "github.com/ethersphere/bee/v2/pkg/storer/internal/events" "github.com/ethersphere/bee/v2/pkg/storer/internal/reserve" @@ -48,3 +50,24 @@ func DefaultOptions() *Options { func TransformedAddress(hasher bmt.Hasher, ch swarm.Chunk, chType swarm.ChunkType) (swarm.Address, error) { return transformedAddress(hasher, ch.Address(), ch.Data(), chType) } + +const DirectUploadRequeueLimit = directUploadRequeueLimit + +// Requeue exposes requeue to tests with chunks only. +type Requeue struct{ q *requeue } + +func NewRequeue(limit int) *Requeue { return &Requeue{q: newRequeue(limit)} } + +func (r *Requeue) Add(ch swarm.Chunk) (swarm.Chunk, bool) { + oldest, full := r.q.add(requeued{ctx: context.Background(), ch: ch}) + return oldest.ch, full +} + +func (r *Requeue) Drain() []swarm.Chunk { + items := r.q.drain() + chunks := make([]swarm.Chunk, len(items)) + for i, it := range items { + chunks[i] = it.ch + } + return chunks +} diff --git a/pkg/storer/metrics.go b/pkg/storer/metrics.go index 787e8a7c9aa..5ae0f614ea1 100644 --- a/pkg/storer/metrics.go +++ b/pkg/storer/metrics.go @@ -36,6 +36,8 @@ type metrics struct { ReserveSampleRunSummary *prometheus.GaugeVec ReserveSampleLastRunTimestamp prometheus.Gauge RecoveryPrunedChunkCount prometheus.Counter + DirectUploadRequeued prometheus.Counter + DirectUploadRequeueFailed prometheus.Counter } // newMetrics is a convenient constructor for creating new metrics. @@ -201,6 +203,22 @@ func newMetrics() metrics { Help: "Number of corrupted chunks pruned from the index during sharky recovery.", }, ), + DirectUploadRequeued: prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: m.Namespace, + Subsystem: subsystem, + Name: "direct_upload_requeued", + Help: "Number of direct upload chunks that failed once and wait for one more push at the end of the upload.", + }, + ), + DirectUploadRequeueFailed: prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: m.Namespace, + Subsystem: subsystem, + Name: "direct_upload_requeue_failed", + Help: "Number of requeued direct upload chunks that failed their second push, failing the upload.", + }, + ), } } diff --git a/pkg/storer/netstore.go b/pkg/storer/netstore.go index 5be414a8395..34f0691615e 100644 --- a/pkg/storer/netstore.go +++ b/pkg/storer/netstore.go @@ -30,6 +30,10 @@ func (db *DB) DirectUpload() PutterSession { // already encountered error. eg, egCtx := errgroup.WithContext(context.Background()) + // chunks that failed once wait here for one more push at the end of the + // upload + q := newRequeue(directUploadRequeueLimit) + return &putterSession{ Putter: putterWithMetrics{ storage.PutterFunc(func(ctx context.Context, ch swarm.Chunk) error { @@ -43,67 +47,123 @@ func (db *DB) DirectUpload() PutterSession { case <-egCtx.Done(): return context.Cause(egCtx) } - eg.Go(safe.RunFunc(db.logger, "storer-netstore-direct-upload", func() (err error) { + eg.Go(safe.RunFunc(db.logger, "storer-netstore-direct-upload", func() error { defer func() { <-db.directUploadLimiter }() - span, logger, ctx := db.tracer.FollowSpanFromContext(ctx, "put-direct-upload", db.logger) - span.SetAttributes(attribute.String("swarm.chunk.address", ch.Address().String())) - defer func() { - if err != nil { - tracing.RecordError(span, err) - } - span.End() - }() + err := db.pushChunk(ctx, egCtx, ch) + if !errors.Is(err, pusher.ErrCouldNotSync) { + return err + } + db.metrics.DirectUploadRequeued.Inc() + db.logger.Debug("direct upload: chunk failed, will push it again at the end of the upload", "chunk_address", ch.Address()) - for { - op := &pusher.Op{Chunk: ch, Err: make(chan error, 1), Direct: true, Span: span} - select { - case <-ctx.Done(): - return ctx.Err() - case <-egCtx.Done(): - return egCtx.Err() - case <-db.quit: - return ErrDBQuit - case db.pusherFeed <- op: - select { - case <-ctx.Done(): - return ctx.Err() - case <-egCtx.Done(): - return egCtx.Err() - case <-db.quit: - return ErrDBQuit - case err := <-op.Err: - if errors.Is(err, pushsync.ErrShallowReceipt) { - logger.Debug("direct upload: shallow receipt received, retrying", "chunk", ch.Address()) - } else if errors.Is(err, topology.ErrNotFound) { - logger.Debug("direct upload: no peers available, retrying", "chunk", ch.Address()) - // wait for peers instead of resubmitting in a tight loop - select { - case <-time.After(directUploadRetryDelay): - case <-ctx.Done(): - return ctx.Err() - case <-egCtx.Done(): - return egCtx.Err() - case <-db.quit: - return ErrDBQuit - } - } else { - return err - } - } - } + oldest, full := q.add(requeued{ctx: ctx, ch: ch}) + if !full { + return nil } + return db.pushRequeued(egCtx, oldest) })) return nil }), db.metrics, "netstore", }, - done: func(swarm.Address) error { return eg.Wait() }, + done: func(swarm.Address) error { + if err := eg.Wait(); err != nil { + return err + } + items := q.drain() + if len(items) == 0 { + return nil + } + rg, rgCtx := errgroup.WithContext(context.Background()) + for _, r := range items { + select { + case db.directUploadLimiter <- struct{}{}: + case <-rgCtx.Done(): + return rg.Wait() + } + rg.Go(safe.RunFunc(db.logger, "storer-netstore-direct-upload-requeued", func() error { + defer func() { <-db.directUploadLimiter }() + return db.pushRequeued(rgCtx, r) + })) + } + return rg.Wait() + }, cleanup: func() error { _ = eg.Wait(); return nil }, } } +// pushRequeued pushes a chunk that failed once for the second and last time. +func (db *DB) pushRequeued(egCtx context.Context, r requeued) error { + err := db.pushChunk(r.ctx, egCtx, r.ch) + if err != nil { + db.metrics.DirectUploadRequeueFailed.Inc() + db.logger.Debug("direct upload: chunk failed again", "chunk_address", r.ch.Address(), "error", err) + } + return err +} + +// pushChunk pushes one chunk of a direct upload until it is synced or fails +// with an error a resubmit cannot fix. +func (db *DB) pushChunk(ctx, egCtx context.Context, ch swarm.Chunk) (err error) { + span, logger, ctx := db.tracer.FollowSpanFromContext(ctx, "put-direct-upload", db.logger) + span.SetAttributes(attribute.String("swarm.chunk.address", ch.Address().String())) + defer func() { + if err != nil { + tracing.RecordError(span, err) + } + span.End() + }() + + for { + // the select below picks at random among ready cases, so check the + // contexts first: a gone client must not get another push + if err := ctx.Err(); err != nil { + return err + } + if err := egCtx.Err(); err != nil { + return err + } + op := &pusher.Op{Chunk: ch, Err: make(chan error, 1), Direct: true, Span: span} + select { + case <-ctx.Done(): + return ctx.Err() + case <-egCtx.Done(): + return egCtx.Err() + case <-db.quit: + return ErrDBQuit + case db.pusherFeed <- op: + select { + case <-ctx.Done(): + return ctx.Err() + case <-egCtx.Done(): + return egCtx.Err() + case <-db.quit: + return ErrDBQuit + case err := <-op.Err: + if errors.Is(err, pushsync.ErrShallowReceipt) { + logger.Debug("direct upload: shallow receipt received, retrying", "chunk", ch.Address()) + } else if errors.Is(err, topology.ErrNotFound) { + logger.Debug("direct upload: no peers available, retrying", "chunk", ch.Address()) + // wait for peers instead of resubmitting in a tight loop + select { + case <-time.After(directUploadRetryDelay): + case <-ctx.Done(): + return ctx.Err() + case <-egCtx.Done(): + return egCtx.Err() + case <-db.quit: + return ErrDBQuit + } + } else { + return err + } + } + } + } +} + // Download is the implementation of the NetStore.Download method. func (db *DB) Download(cache bool) storage.Getter { return getterWithMetrics{ diff --git a/pkg/storer/netstore_test.go b/pkg/storer/netstore_test.go index 59940c8ce2e..af57231785e 100644 --- a/pkg/storer/netstore_test.go +++ b/pkg/storer/netstore_test.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "slices" + "sync" "testing" "testing/synctest" "time" @@ -447,3 +448,254 @@ func TestDirectUploadStopsOnError(t *testing.T) { } }) } + +// scriptedPusher answers direct pushes from a script, by chunk address and +// the number of times that chunk has been pushed, and records the pushes. +type scriptedPusher struct { + mu sync.Mutex + pushes []swarm.Address + answer func(addr swarm.Address, n int) error +} + +func (p *scriptedPusher) run(feed <-chan *pusher.Op, quit <-chan struct{}) { + for { + select { + case op := <-feed: + p.mu.Lock() + p.pushes = append(p.pushes, op.Chunk.Address()) + n := p.countLocked(op.Chunk.Address()) + p.mu.Unlock() + op.Err <- p.answer(op.Chunk.Address(), n) + case <-quit: + return + } + } +} + +func (p *scriptedPusher) countLocked(addr swarm.Address) int { + n := 0 + for _, a := range p.pushes { + if a.Equal(addr) { + n++ + } + } + return n +} + +func (p *scriptedPusher) count(addr swarm.Address) int { + p.mu.Lock() + defer p.mu.Unlock() + return p.countLocked(addr) +} + +func (p *scriptedPusher) order() []swarm.Address { + p.mu.Lock() + defer p.mu.Unlock() + return slices.Clone(p.pushes) +} + +// withScriptedPusher runs f against an in-memory storer whose direct pushes are +// answered by answer. It must be called inside a synctest bubble. +func withScriptedPusher(t *testing.T, answer func(addr swarm.Address, n int) error, f func(lstore *storer.DB, p *scriptedPusher)) { + t.Helper() + + lstore, err := storer.New(context.Background(), "", dbTestOps(swarm.RandAddress(t), 0, nil, nil, time.Second)) + if err != nil { + t.Fatal(err) + } + + p := &scriptedPusher{answer: answer} + quit := make(chan struct{}) + defer func() { + close(quit) + if err := lstore.Close(); err != nil { + t.Errorf("close: %v", err) + } + // leveldb waits up to a second for its memory pool on close + time.Sleep(time.Second) + }() + go p.run(lstore.PusherFeed(), quit) + + f(lstore, p) +} + +// TestDirectUploadRequeue verifies that a chunk which fails a direct upload is +// pushed once more at the end of the upload before the upload fails. +func TestDirectUploadRequeue(t *testing.T) { + t.Parallel() + + // failFirst fails the first push of every chunk in fail and any push of + // a chunk in failAlways. + failFirst := func(fail, failAlways []swarm.Chunk) func(swarm.Address, int) error { + return func(addr swarm.Address, n int) error { + for _, ch := range failAlways { + if ch.Address().Equal(addr) { + return pusher.ErrCouldNotSync + } + } + for _, ch := range fail { + if ch.Address().Equal(addr) && n == 1 { + return pusher.ErrCouldNotSync + } + } + return nil + } + } + + put := func(t *testing.T, session storer.PutterSession, ch swarm.Chunk) { + t.Helper() + if err := session.Put(context.Background(), ch); err != nil { + t.Fatalf("session.Put(...): unexpected error: %v", err) + } + synctest.Wait() + } + + t.Run("retried chunk succeeds", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + chunks := chunktesting.GenerateTestRandomChunks(2) + withScriptedPusher(t, failFirst(chunks[:1], nil), func(lstore *storer.DB, p *scriptedPusher) { + session := lstore.DirectUpload() + put(t, session, chunks[0]) + put(t, session, chunks[1]) + + if err := session.Done(swarm.ZeroAddress); err != nil { + t.Fatalf("session.Done(): unexpected error: %v", err) + } + want := []swarm.Address{chunks[0].Address(), chunks[1].Address(), chunks[0].Address()} + if got := p.order(); !slices.EqualFunc(got, want, swarm.Address.Equal) { + t.Fatalf("pushes %v, want %v", got, want) + } + }) + }) + }) + + t.Run("fails twice", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + chunks := chunktesting.GenerateTestRandomChunks(1) + withScriptedPusher(t, failFirst(nil, chunks), func(lstore *storer.DB, p *scriptedPusher) { + session := lstore.DirectUpload() + put(t, session, chunks[0]) + + if err := session.Done(swarm.ZeroAddress); !errors.Is(err, pusher.ErrCouldNotSync) { + t.Fatalf("session.Done(): got %v, want %v", err, pusher.ErrCouldNotSync) + } + if got := p.count(chunks[0].Address()); got != 2 { + t.Fatalf("got %d pushes, want 2", got) + } + }) + }) + }) + + t.Run("full queue retries oldest first", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + chunks := chunktesting.GenerateTestRandomChunks(storer.DirectUploadRequeueLimit + 1) + withScriptedPusher(t, failFirst(chunks, nil), func(lstore *storer.DB, p *scriptedPusher) { + session := lstore.DirectUpload() + for _, ch := range chunks { + put(t, session, ch) + } + + if got := p.count(chunks[0].Address()); got != 2 { + t.Fatalf("oldest chunk: got %d pushes before done, want 2", got) + } + if err := session.Done(swarm.ZeroAddress); err != nil { + t.Fatalf("session.Done(): unexpected error: %v", err) + } + for i, ch := range chunks { + if got := p.count(ch.Address()); got != 2 { + t.Fatalf("chunk %d: got %d pushes, want 2", i, got) + } + } + }) + }) + }) + + t.Run("full queue oldest fails again", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + chunks := chunktesting.GenerateTestRandomChunks(storer.DirectUploadRequeueLimit + 1) + withScriptedPusher(t, failFirst(chunks, chunks[:1]), func(lstore *storer.DB, p *scriptedPusher) { + session := lstore.DirectUpload() + for _, ch := range chunks { + put(t, session, ch) + } + + if err := session.Done(swarm.ZeroAddress); !errors.Is(err, pusher.ErrCouldNotSync) { + t.Fatalf("session.Done(): got %v, want %v", err, pusher.ErrCouldNotSync) + } + if got := p.count(chunks[0].Address()); got != 2 { + t.Fatalf("oldest chunk: got %d pushes, want 2", got) + } + }) + }) + }) + + t.Run("first pass error skips retry", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + chunks := chunktesting.GenerateTestRandomChunks(2) + answer := func(addr swarm.Address, n int) error { + if addr.Equal(chunks[0].Address()) { + return pusher.ErrCouldNotSync + } + return pushsync.ErrNoPush + } + withScriptedPusher(t, answer, func(lstore *storer.DB, p *scriptedPusher) { + session := lstore.DirectUpload() + put(t, session, chunks[0]) + put(t, session, chunks[1]) + + if err := session.Done(swarm.ZeroAddress); !errors.Is(err, pushsync.ErrNoPush) { + t.Fatalf("session.Done(): got %v, want %v", err, pushsync.ErrNoPush) + } + if got := p.count(chunks[0].Address()); got != 1 { + t.Fatalf("got %d pushes, want 1", got) + } + }) + }) + }) + + t.Run("cleanup does not retry", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + chunks := chunktesting.GenerateTestRandomChunks(1) + withScriptedPusher(t, failFirst(chunks, nil), func(lstore *storer.DB, p *scriptedPusher) { + session := lstore.DirectUpload() + put(t, session, chunks[0]) + + if err := session.Cleanup(); err != nil { + t.Fatalf("session.Cleanup(): unexpected error: %v", err) + } + if got := p.count(chunks[0].Address()); got != 1 { + t.Fatalf("got %d pushes, want 1", got) + } + }) + }) + }) + + t.Run("client gone before done", func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + chunks := chunktesting.GenerateTestRandomChunks(1) + withScriptedPusher(t, failFirst(chunks, nil), func(lstore *storer.DB, p *scriptedPusher) { + ctx, cancel := context.WithCancel(context.Background()) + session := lstore.DirectUpload() + if err := session.Put(ctx, chunks[0]); err != nil { + t.Fatalf("session.Put(...): unexpected error: %v", err) + } + synctest.Wait() + cancel() + + if err := session.Done(swarm.ZeroAddress); !errors.Is(err, context.Canceled) { + t.Fatalf("session.Done(): got %v, want %v", err, context.Canceled) + } + if got := p.count(chunks[0].Address()); got != 1 { + t.Fatalf("got %d pushes, want 1", got) + } + }) + }) + }) +} diff --git a/pkg/storer/requeue.go b/pkg/storer/requeue.go new file mode 100644 index 00000000000..9af1dec79ef --- /dev/null +++ b/pkg/storer/requeue.go @@ -0,0 +1,60 @@ +// 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 storer + +import ( + "context" + "sync" + + "github.com/ethersphere/bee/v2/pkg/swarm" +) + +// directUploadRequeueLimit is how many chunks of one direct upload may wait +// for their second push at the same time. Many failures in a short time point +// to a problem that a later push will not fix, so the upload then fails fast. +const directUploadRequeueLimit = 16 + +// requeued is a chunk of a direct upload waiting for its second push. +type requeued struct { + ctx context.Context // the Put context, still alive at Done + ch swarm.Chunk +} + +// requeue holds the chunks of a direct upload that failed once. They are +// pushed again at the end of the upload. +type requeue struct { + mu sync.Mutex + limit int + items []requeued +} + +func newRequeue(limit int) *requeue { + return &requeue{limit: limit} +} + +// add appends r. When the queue is full, it removes the oldest entry in the +// same step and returns it; the caller must push that one now. +func (q *requeue) add(r requeued) (requeued, bool) { + q.mu.Lock() + defer q.mu.Unlock() + + if len(q.items) < q.limit { + q.items = append(q.items, r) + return requeued{}, false + } + oldest := q.items[0] + q.items = append(q.items[1:], r) + return oldest, true +} + +// drain returns the entries, oldest first, and empties the queue. +func (q *requeue) drain() []requeued { + q.mu.Lock() + defer q.mu.Unlock() + + items := q.items + q.items = nil + return items +} diff --git a/pkg/storer/requeue_test.go b/pkg/storer/requeue_test.go new file mode 100644 index 00000000000..60ab94b795a --- /dev/null +++ b/pkg/storer/requeue_test.go @@ -0,0 +1,88 @@ +// 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 storer_test + +import ( + "sync" + "testing" + + chunktesting "github.com/ethersphere/bee/v2/pkg/storage/testing" + "github.com/ethersphere/bee/v2/pkg/storer" + "github.com/ethersphere/bee/v2/pkg/swarm" +) + +func TestRequeueAdd(t *testing.T) { + t.Parallel() + + chunks := chunktesting.GenerateTestRandomChunks(3) + q := storer.NewRequeue(2) + + if _, full := q.Add(chunks[0]); full { + t.Fatal("add 1: queue reported full") + } + if _, full := q.Add(chunks[1]); full { + t.Fatal("add 2: queue reported full") + } + oldest, full := q.Add(chunks[2]) + if !full { + t.Fatal("add 3: queue not reported full") + } + if !oldest.Address().Equal(chunks[0].Address()) { + t.Fatalf("add 3: got oldest %s, want %s", oldest.Address(), chunks[0].Address()) + } + + got := q.Drain() + if len(got) != 2 || !got[0].Address().Equal(chunks[1].Address()) || !got[1].Address().Equal(chunks[2].Address()) { + t.Fatalf("drain: got %v, want [%s %s]", addresses(got), chunks[1].Address(), chunks[2].Address()) + } + if got := q.Drain(); len(got) != 0 { + t.Fatalf("second drain: got %d entries, want 0", len(got)) + } +} + +func TestRequeueConcurrentAdd(t *testing.T) { + t.Parallel() + + limit := storer.DirectUploadRequeueLimit + q := storer.NewRequeue(limit) + for _, ch := range chunktesting.GenerateTestRandomChunks(limit) { + q.Add(ch) + } + + added := chunktesting.GenerateTestRandomChunks(64) + returned := make([]swarm.Chunk, len(added)) + fulls := make([]bool, len(added)) + + var wg sync.WaitGroup + for i, ch := range added { + wg.Go(func() { + returned[i], fulls[i] = q.Add(ch) + }) + } + wg.Wait() + + seen := make(map[string]bool) + for i := range added { + if !fulls[i] { + t.Fatalf("add %d: queue not reported full", i) + } + key := returned[i].Address().ByteString() + if seen[key] { + t.Fatalf("add %d: chunk %s returned twice", i, returned[i].Address()) + } + seen[key] = true + } + if got := len(q.Drain()); got != limit { + t.Fatalf("drain: got %d entries, want %d", got, limit) + } +} + +func addresses(chunks []swarm.Chunk) []swarm.Address { + addrs := make([]swarm.Address, len(chunks)) + for i, ch := range chunks { + addrs[i] = ch.Address() + } + return addrs +} From 108bf5040919f58ed6f7bea9b96a84c725081a13 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 14:42:25 +0200 Subject: [PATCH 05/11] fix(pusher): park a failed deferred chunk once before giving up on it --- pkg/pusher/export_test.go | 28 +++++ pkg/pusher/metrics.go | 7 ++ pkg/pusher/parked.go | 90 ++++++++++++++ pkg/pusher/parked_test.go | 53 ++++++++ pkg/pusher/pusher.go | 43 +++++-- pkg/pusher/pusher_test.go | 230 +++++++++++++++++++++++++++++------ pkg/storer/subscribe_push.go | 7 ++ 7 files changed, 411 insertions(+), 47 deletions(-) create mode 100644 pkg/pusher/export_test.go create mode 100644 pkg/pusher/parked.go create mode 100644 pkg/pusher/parked_test.go diff --git a/pkg/pusher/export_test.go b/pkg/pusher/export_test.go new file mode 100644 index 00000000000..fe8c48f6295 --- /dev/null +++ b/pkg/pusher/export_test.go @@ -0,0 +1,28 @@ +// 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 pusher + +import ( + "time" + + "github.com/ethersphere/bee/v2/pkg/swarm" +) + +const ParkDelay = parkDelay + +// Parked exposes parked to tests. +type Parked struct{ p *parked } + +func NewParked(delay, ttl time.Duration) *Parked { return &Parked{p: newParked(delay, ttl)} } + +func (p *Parked) Park(idAddress swarm.Address, batchID []byte, now time.Time) bool { + return p.p.park(idAddress, batchID, now) +} + +func (p *Parked) Waiting(idAddress swarm.Address, batchID []byte, now time.Time) bool { + return p.p.waiting(idAddress, batchID, now) +} + +func (p *Parked) Forget(idAddress swarm.Address, batchID []byte) { p.p.forget(idAddress, batchID) } diff --git a/pkg/pusher/metrics.go b/pkg/pusher/metrics.go index d0000cce13b..0eb3a2704f4 100644 --- a/pkg/pusher/metrics.go +++ b/pkg/pusher/metrics.go @@ -13,6 +13,7 @@ type metrics struct { TotalToPush prometheus.Counter TotalSynced prometheus.Counter TotalCouldNotSync prometheus.Counter + TotalParked prometheus.Counter TotalErrors prometheus.Counter MarkAndSweepTime prometheus.Histogram SyncTime prometheus.Histogram @@ -41,6 +42,12 @@ func newMetrics() metrics { Name: "total_could_not_sync", Help: "Total chunks abandoned after exhausting retries with no valid receipt (shallow receipt or no peer in the correct neighbourhood).", }), + TotalParked: prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: m.Namespace, + Subsystem: subsystem, + Name: "total_parked", + Help: "Total deferred chunks left in the upload queue for one more round after they failed to be placed.", + }), TotalErrors: prometheus.NewCounter(prometheus.CounterOpts{ Namespace: m.Namespace, Subsystem: subsystem, diff --git a/pkg/pusher/parked.go b/pkg/pusher/parked.go new file mode 100644 index 00000000000..85c8055ba81 --- /dev/null +++ b/pkg/pusher/parked.go @@ -0,0 +1,90 @@ +// 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 pusher + +import ( + "sync" + "time" + + "github.com/ethersphere/bee/v2/pkg/swarm" +) + +const ( + // parkDelay is how long a deferred chunk that failed its first round + // stays untouched in the upload queue. Pushsync skips the peers that + // refused the chunk for as long (skiplistDur), so an earlier round would + // end the same way. + parkDelay = 5 * time.Minute + + // parkTTL is how long after its delay a parked entry is kept for a + // chunk that never came back, e.g. because its upload was removed. + parkTTL = time.Hour +) + +// parked tracks deferred chunks that failed their first round and wait in the +// upload queue for their second and last one. +type parked struct { + mu sync.Mutex + entries map[[64]byte]time.Time // identity address + batch ID -> until + delay time.Duration + ttl time.Duration +} + +func newParked(delay, ttl time.Duration) *parked { + return &parked{ + entries: make(map[[64]byte]time.Time), + delay: delay, + ttl: ttl, + } +} + +func parkedKey(idAddress swarm.Address, batchID []byte) [64]byte { + var key [64]byte + copy(key[:32], idAddress.Bytes()) + copy(key[32:], batchID) + return key +} + +// park parks the chunk for its second round. It reports false when the chunk +// was parked before; the caller then gives up on it. +func (p *parked) park(idAddress swarm.Address, batchID []byte, now time.Time) bool { + key := parkedKey(idAddress, batchID) + + p.mu.Lock() + defer p.mu.Unlock() + + for k, until := range p.entries { + if now.After(until.Add(p.ttl)) { + delete(p.entries, k) + } + } + + if _, ok := p.entries[key]; ok { + return false + } + p.entries[key] = now.Add(p.delay) + return true +} + +// waiting reports whether the chunk must still be left alone. The entry stays +// after its delay, so that park recognises the second round. +func (p *parked) waiting(idAddress swarm.Address, batchID []byte, now time.Time) bool { + key := parkedKey(idAddress, batchID) + + p.mu.Lock() + defer p.mu.Unlock() + + until, ok := p.entries[key] + return ok && now.Before(until) +} + +// forget removes the chunk once it is synced, stored or given up on. +func (p *parked) forget(idAddress swarm.Address, batchID []byte) { + key := parkedKey(idAddress, batchID) + + p.mu.Lock() + delete(p.entries, key) + p.mu.Unlock() +} diff --git a/pkg/pusher/parked_test.go b/pkg/pusher/parked_test.go new file mode 100644 index 00000000000..da7cdeef34a --- /dev/null +++ b/pkg/pusher/parked_test.go @@ -0,0 +1,53 @@ +// 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 pusher_test + +import ( + "testing" + "time" + + "github.com/ethersphere/bee/v2/pkg/pusher" + "github.com/ethersphere/bee/v2/pkg/swarm" +) + +func TestParked(t *testing.T) { + t.Parallel() + + const ( + delay = 5 * time.Minute + ttl = time.Hour + ) + t0 := time.Date(2026, 10, 19, 0, 0, 0, 0, time.UTC) + batch := []byte("batch") + a, b := swarm.RandAddress(t), swarm.RandAddress(t) + + p := pusher.NewParked(delay, ttl) + + if !p.Park(a, batch, t0) { + t.Fatal("first park: got false, want true") + } + if !p.Waiting(a, batch, t0.Add(time.Minute)) { + t.Fatal("before the delay: not waiting") + } + if p.Waiting(a, batch, t0.Add(delay)) { + t.Fatal("after the delay: still waiting") + } + if p.Park(a, batch, t0.Add(6*time.Minute)) { + t.Fatal("second park: got true, want false") + } + + p.Forget(a, batch) + if !p.Park(a, batch, t0) { + t.Fatal("park after forget: got false, want true") + } + + // an entry left alone is dropped once it is older than the ttl + if !p.Park(b, batch, t0.Add(delay+ttl+time.Second)) { + t.Fatal("park of another chunk: got false, want true") + } + if !p.Park(a, batch, t0.Add(delay+ttl+time.Second)) { + t.Fatal("park after the ttl: got false, want true") + } +} diff --git a/pkg/pusher/pusher.go b/pkg/pusher/pusher.go index a879c45bc2c..c68c82b05e3 100644 --- a/pkg/pusher/pusher.go +++ b/pkg/pusher/pusher.go @@ -61,6 +61,7 @@ type Service struct { chunksWorkerQuitC chan struct{} inflight *inflight attempts *attempts + parked *parked smuggler chan OpChan } @@ -94,6 +95,7 @@ func New( chunksWorkerQuitC: make(chan struct{}), inflight: newInflight(), attempts: &attempts{retryCount: retryCount, attempts: make(map[string]int)}, + parked: newParked(parkDelay, parkTTL), smuggler: make(chan OpChan), } go p.chunksWorker(startupStabilizer) @@ -238,6 +240,9 @@ func (s *Service) chunksWorker(startupStabilizer stabilization.Subscriber) { continue } op.identityAddress = idAddress + if !op.Direct && s.parked.waiting(idAddress, op.Chunk.Stamp().BatchID(), time.Now()) { + continue + } if s.inflight.set(idAddress, op.Chunk.Stamp().BatchID()) { if op.Direct { select { @@ -276,6 +281,7 @@ func (s *Service) pushDeferred(ctx context.Context, logger log.Logger, op *Op) ( "chunk_address", op.Chunk.Address(), "error", err, ) + s.parked.forget(op.identityAddress, op.Chunk.Stamp().BatchID()) return false, errors.Join(err, s.storer.Report(ctx, op.Chunk, storage.ChunkCouldNotSync)) } @@ -295,16 +301,14 @@ func (s *Service) pushDeferred(ctx context.Context, logger log.Logger, op *Op) ( } s.metrics.TotalSynced.Inc() s.attempts.delete(op.identityAddress) - case errors.Is(err, pushsync.ErrShallowReceipt), errors.Is(err, pushsync.ErrOutOfDepthStoring): + s.parked.forget(op.identityAddress, op.Chunk.Stamp().BatchID()) + case errors.Is(err, pushsync.ErrOutOfDepthStoring): + return s.parkOrGiveUp(ctx, loggerV1, op, err) + case errors.Is(err, pushsync.ErrShallowReceipt): if s.shallowReceipt(op.identityAddress) { return true, err } - // budget exhausted; report CouldNotSync so the chunk isn't marked delivered - s.metrics.TotalCouldNotSync.Inc() - if err := s.storer.Report(ctx, op.Chunk, storage.ChunkCouldNotSync); err != nil { - loggerV1.Error(err, "pusher: failed to report sync status") - return true, err - } + return s.parkOrGiveUp(ctx, loggerV1, op, err) case err == nil: if err := s.storer.Report(ctx, op.Chunk, storage.ChunkSynced); err != nil { loggerV1.Error(err, "pusher: failed to report sync status") @@ -312,6 +316,7 @@ func (s *Service) pushDeferred(ctx context.Context, logger log.Logger, op *Op) ( } s.metrics.TotalSynced.Inc() s.attempts.delete(op.identityAddress) + s.parked.forget(op.identityAddress, op.Chunk.Stamp().BatchID()) default: loggerV1.Error(err, "pusher: failed PushChunkToClosest") return true, err @@ -320,6 +325,30 @@ func (s *Service) pushDeferred(ctx context.Context, logger log.Logger, op *Op) ( return false, nil } +// parkOrGiveUp handles a deferred chunk that no node within its neighborhood +// took, either at once or after its shallow receipt budget was spent. The +// first time, the chunk stays in the upload queue and gets one more round +// after parkDelay, with a fresh budget. The second time, it is given up on. +func (s *Service) parkOrGiveUp(ctx context.Context, logger log.Logger, op *Op, err error) (bool, error) { + batchID := op.Chunk.Stamp().BatchID() + s.attempts.delete(op.identityAddress) + + if s.parked.park(op.identityAddress, batchID, time.Now()) { + s.metrics.TotalParked.Inc() + logger.Debug("chunk not placed, will retry later", "chunk_address", op.Chunk.Address(), "retry_in", parkDelay, "error", err) + return false, err + } + + s.parked.forget(op.identityAddress, batchID) + // report CouldNotSync so the chunk isn't marked delivered + s.metrics.TotalCouldNotSync.Inc() + if err := s.storer.Report(ctx, op.Chunk, storage.ChunkCouldNotSync); err != nil { + logger.Error(err, "pusher: failed to report sync status") + return true, err + } + return false, nil +} + func (s *Service) pushDirect(ctx context.Context, logger log.Logger, op *Op) error { loggerV1 := logger.V(1).Build() diff --git a/pkg/pusher/pusher_test.go b/pkg/pusher/pusher_test.go index 755d16dea8b..8e51f299446 100644 --- a/pkg/pusher/pusher_test.go +++ b/pkg/pusher/pusher_test.go @@ -11,6 +11,7 @@ import ( "sync" "sync/atomic" "testing" + "testing/synctest" "time" "github.com/ethereum/go-ethereum/common" @@ -372,28 +373,37 @@ func TestShallowReceiptExhausted(t *testing.T) { return &pushsync.Receipt{Address: chunk.Address()}, pushsync.ErrShallowReceipt }) + // a deferred chunk gets one more round after the park delay t.Run("deferred", func(t *testing.T) { t.Parallel() - storer := &reportingStorer{ - mockStorer: &mockStorer{chunks: make(chan swarm.Chunk)}, - reported: make(chan storage.ChunkState, 1), - } - _ = createPusher(t, storer, pushSyncService, defaultMockBatchStore, defaultRetryCount) + synctest.Test(t, func(t *testing.T) { + var calls atomic.Int32 + pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { + calls.Add(1) + return &pushsync.Receipt{Address: chunk.Address()}, pushsync.ErrShallowReceipt + }) + storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) + + offer(storer, chunk) + if got := int(calls.Load()); got != defaultRetryCount { + t.Fatalf("round 1: got %d push attempts, want %d", got, defaultRetryCount) + } + expectNoReport(t, storer) - before := callCount.Load() - chunk := testingc.GenerateTestRandomChunk() - storer.chunks <- chunk + // offered again while parked: skipped + offer(storer, chunk) + if got := int(calls.Load()); got != defaultRetryCount { + t.Fatalf("while parked: got %d push attempts, want %d", got, defaultRetryCount) + } - if state := <-storer.reported; state != storage.ChunkCouldNotSync { - t.Fatalf("got reported state %v, want %v", state, storage.ChunkCouldNotSync) - } - if storer.isReported(chunk, storage.ChunkSynced) { - t.Fatal("chunk reported as synced") - } - if got := int(callCount.Load() - before); got < defaultRetryCount { - t.Fatalf("got %d push attempts, want at least %d", got, defaultRetryCount) - } + time.Sleep(pusher.ParkDelay) + offer(storer, chunk) + if got := int(calls.Load()); got != 2*defaultRetryCount { + t.Fatalf("round 2: got %d push attempts, want %d", got, 2*defaultRetryCount) + } + expectReport(t, storer, storage.ChunkCouldNotSync) + }) }) t.Run("direct", func(t *testing.T) { @@ -430,40 +440,50 @@ func TestShallowReceiptExhausted(t *testing.T) { }) } -// TestOutOfDepthStoring tests that a chunk which no reachable node is -// responsible for is given up on instead of being retried forever. +// TestOutOfDepthStoring tests what happens to a chunk which no reachable node +// is responsible for. func TestOutOfDepthStoring(t *testing.T) { t.Parallel() - var callCount atomic.Int32 - pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { - callCount.Add(1) - return nil, pushsync.ErrOutOfDepthStoring - }) - + // a deferred chunk stays in the upload queue for one more round after the + // park delay, then it is given up on t.Run("deferred", func(t *testing.T) { t.Parallel() - storer := &reportingStorer{ - mockStorer: &mockStorer{chunks: make(chan swarm.Chunk)}, - reported: make(chan storage.ChunkState, 1), - } - _ = createPusher(t, storer, pushSyncService, defaultMockBatchStore, defaultRetryCount) - - chunk := testingc.GenerateTestRandomChunk() - storer.chunks <- chunk + synctest.Test(t, func(t *testing.T) { + var calls atomic.Int32 + pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { + calls.Add(1) + return nil, pushsync.ErrOutOfDepthStoring + }) + storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) + + // the queue offers the chunk again and again while it is pending + for range 3 { + offer(storer, chunk) + } + if got := calls.Load(); got != 1 { + t.Fatalf("got %d push attempts before the park delay, want 1", got) + } + expectNoReport(t, storer) - if state := <-storer.reported; state != storage.ChunkCouldNotSync { - t.Fatalf("got reported state %v, want %v", state, storage.ChunkCouldNotSync) - } - if storer.isReported(chunk, storage.ChunkSynced) || storer.isReported(chunk, storage.ChunkStored) { - t.Fatal("chunk reported as synced or stored") - } + time.Sleep(pusher.ParkDelay) + offer(storer, chunk) + if got := calls.Load(); got != 2 { + t.Fatalf("got %d push attempts after the park delay, want 2", got) + } + expectReport(t, storer, storage.ChunkCouldNotSync) + }) }) + // a direct upload has no queue to wait in, so it fails right away t.Run("direct", func(t *testing.T) { t.Parallel() + pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { + return nil, pushsync.ErrOutOfDepthStoring + }) + storer := &mockStorer{chunks: make(chan swarm.Chunk)} pusherSvc := createPusher(t, storer, pushSyncService, defaultMockBatchStore, defaultRetryCount) @@ -473,7 +493,6 @@ func TestOutOfDepthStoring(t *testing.T) { errC := make(chan error, 1) newFeed <- &pusher.Op{Chunk: testingc.GenerateTestRandomChunk(), Err: errC, Direct: true} - // the first attempt already ends the upload with a terminal error if err := <-errC; !errors.Is(err, pusher.ErrCouldNotSync) { t.Fatalf("got %v, want %v", err, pusher.ErrCouldNotSync) } @@ -938,3 +957,134 @@ func TestDirectBatchGone(t *testing.T) { t.Fatalf("got %d pushes, want 0", got) } } + +// newParkingPusher starts a pusher over a storer whose push queue the test +// drives with offer. It must be called inside a synctest bubble. +func newParkingPusher(t *testing.T, pushSyncService pushsync.PushSyncer, batchExist postage.BatchExist) (*reportingStorer, swarm.Chunk) { + t.Helper() + + storer := &reportingStorer{ + mockStorer: &mockStorer{chunks: make(chan swarm.Chunk)}, + reported: make(chan storage.ChunkState, 8), + } + pusherSvc := pusher.New(1, storer, pushSyncService, batchExist, log.Noop, stabilmock.NewSubscriber(true), defaultRetryCount) + t.Cleanup(func() { _ = pusherSvc.Close() }) + + return storer, testingc.GenerateTestRandomChunk() +} + +// offer hands the chunk to the pusher as the upload store iteration does and +// waits until the pusher is idle. +func offer(storer *reportingStorer, chunk swarm.Chunk) { + storer.chunks <- chunk + synctest.Wait() +} + +func expectNoReport(t *testing.T, storer *reportingStorer) { + t.Helper() + select { + case state := <-storer.reported: + t.Fatalf("chunk reported as %v, want it left in the queue", state) + default: + } +} + +func expectReport(t *testing.T, storer *reportingStorer, want storage.ChunkState) { + t.Helper() + select { + case state := <-storer.reported: + if state != want { + t.Fatalf("chunk reported as %v, want %v", state, want) + } + default: + t.Fatalf("chunk not reported, want %v", want) + } +} + +// TestParkedChunkSynced tests that a parked chunk which then syncs is +// forgotten, so a later failure of the same chunk parks it again. +func TestParkedChunkSynced(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + var calls atomic.Int32 + pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { + if calls.Add(1) == 2 { + return &pushsync.Receipt{Address: chunk.Address()}, nil + } + return nil, pushsync.ErrOutOfDepthStoring + }) + storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) + + offer(storer, chunk) + expectNoReport(t, storer) + + time.Sleep(pusher.ParkDelay) + offer(storer, chunk) + expectReport(t, storer, storage.ChunkSynced) + + // uploaded again later and out of depth: a new first round + offer(storer, chunk) + expectNoReport(t, storer) + }) +} + +// TestParkedChunkBatchGone tests that a parked chunk whose batch expires is +// given up on and forgotten. +func TestParkedChunkBatchGone(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { + return nil, pushsync.ErrOutOfDepthStoring + }) + var batchGone atomic.Bool + batchExist := batchstoremock.New(batchstoremock.WithExistsFunc(func(b []byte) (bool, error) { + return !batchGone.Load(), nil + })) + storer, chunk := newParkingPusher(t, pushSyncService, batchExist) + + offer(storer, chunk) + expectNoReport(t, storer) + + batchGone.Store(true) + time.Sleep(pusher.ParkDelay) + offer(storer, chunk) + expectReport(t, storer, storage.ChunkCouldNotSync) + + // the same chunk with a valid batch again: a new first round + batchGone.Store(false) + offer(storer, chunk) + expectNoReport(t, storer) + }) +} + +// TestDeferredOutOfDepthResetsAttempts tests that a deferred chunk parked on +// ErrOutOfDepthStoring gets the full retry budget in its second round. +func TestDeferredOutOfDepthResetsAttempts(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + var calls atomic.Int32 + pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { + if calls.Add(1) == 2 { + return nil, pushsync.ErrOutOfDepthStoring + } + return &pushsync.Receipt{Address: chunk.Address()}, pushsync.ErrShallowReceipt + }) + storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) + + offer(storer, chunk) + if got := calls.Load(); got != 2 { + t.Fatalf("round 1: got %d push attempts, want 2", got) + } + expectNoReport(t, storer) + + time.Sleep(pusher.ParkDelay) + offer(storer, chunk) + if got := int(calls.Load()); got != 2+defaultRetryCount { + t.Fatalf("round 2: got %d push attempts in total, want %d", got, 2+defaultRetryCount) + } + expectReport(t, storer, storage.ChunkCouldNotSync) + }) +} diff --git a/pkg/storer/subscribe_push.go b/pkg/storer/subscribe_push.go index dcd4a201e83..d948adcc3b0 100644 --- a/pkg/storer/subscribe_push.go +++ b/pkg/storer/subscribe_push.go @@ -15,6 +15,10 @@ import ( const subscribePushEventKey = "subscribe-push" +// pushRetryInterval is how often the pending chunks are offered to the +// subscriber again when no upload event triggers it. +const pushRetryInterval = time.Minute + func (db *DB) SubscribePush(ctx context.Context) (<-chan swarm.Chunk, func()) { chunks := make(chan swarm.Chunk) @@ -74,6 +78,9 @@ func (db *DB) SubscribePush(ctx context.Context) (<-chan swarm.Chunk, func()) { return case <-trigger: // wait for the next event + case <-time.After(pushRetryInterval): + // chunks the pusher left in the queue to retry later + // are only seen again by iterating } } }) From 1028719ee394c6b4b01509cd37f92ea5eea206aa Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 14:44:48 +0200 Subject: [PATCH 06/11] fix(pushsync): wait for in-flight pushes before an origin gives up --- pkg/pushsync/pushsync.go | 12 +++- pkg/pushsync/pushsync_test.go | 122 ++++++++++++++++++++++++++++++++++ 2 files changed, 132 insertions(+), 2 deletions(-) diff --git a/pkg/pushsync/pushsync.go b/pkg/pushsync/pushsync.go index df80179ad6b..304e9a18bbe 100644 --- a/pkg/pushsync/pushsync.go +++ b/pkg/pushsync/pushsync.go @@ -408,13 +408,21 @@ func (ps *PushSync) pushToClosest(ctx context.Context, ch swarm.Chunk, origin bo skip := skippeers.NewList(0) defer skip.Close() - for sentErrorsLeft > 0 { + // An origin whose budget is spent still waits for the pushes in flight: + // they run on their own timeout and are paid for when their receipt + // arrives, so dropping their result would waste a valid receipt. + for sentErrorsLeft > 0 || (origin && inflight > 0) { select { case <-ctx.Done(): return nil, ErrNoPush case <-preemptiveTicker: - retry() + if sentErrorsLeft > 0 { + retry() + } case <-retryC: + if sentErrorsLeft <= 0 { + continue + } // Origin peers should not store the chunk initially so that the chunk is always forwarded into the network. // If no peer can be found from an origin peer, the origin peer may store the chunk. diff --git a/pkg/pushsync/pushsync_test.go b/pkg/pushsync/pushsync_test.go index f6096fea3cb..8cd3c779501 100644 --- a/pkg/pushsync/pushsync_test.go +++ b/pkg/pushsync/pushsync_test.go @@ -8,6 +8,7 @@ import ( "bytes" "context" "errors" + "slices" "strings" "sync" "sync/atomic" @@ -1404,3 +1405,124 @@ func chanFunc(c chan<- struct{}) func(swarm.Chunk) { c <- struct{}{} } } + +// slowStorerSetup builds an origin whose closest peer stores the chunk and +// answers with a valid receipt only after delay, while enough other peers fail +// at once to spend the origin's whole push budget before that. +func slowStorerSetup(t *testing.T, delay time.Duration) (*pushsync.PushSync, swarm.Chunk, swarm.Address) { + t.Helper() + + const radius = 2 + + key, err := crypto.GenerateSecp256k1Key() + if err != nil { + t.Fatal(err) + } + signer := crypto.NewDefaultSigner(key) + pubKey, err := signer.PublicKey() + if err != nil { + t.Fatal(err) + } + storer, err := crypto.NewOverlayAddress(*pubKey, 1, blockHash.Bytes()) + if err != nil { + t.Fatal(err) + } + chunk := testingc.GenerateValidRandomChunkAt(t, storer, radius) + + psStorer, _ := createPushSyncNodeWithRadius(t, storer, defaultPrices, nil, nil, signer, radius, mock.WithClosestPeerErr(topology.ErrWantSelf)) + spec := psStorer.Protocol() + streams := slices.Clone(spec.StreamSpecs) + for i := range streams { + handler := streams[i].Handler + streams[i].Handler = func(ctx context.Context, p p2p.Peer, s p2p.Stream) error { + time.Sleep(delay) + return handler(ctx, p, s) + } + } + spec.StreamSpecs = streams + + // the storer is in the chunk's AOR, so the origin multiplexes and its + // budget grows by the extra forwards + const failingPeers = 34 + failing := make(map[string]bool, failingPeers) + peers := make([]swarm.Address, 0, failingPeers+1) + peers = append(peers, storer) + for range failingPeers { + b := swarm.RandAddress(t).Bytes() + b[0] = chunk.Address().Bytes()[0] ^ 0x80 // po 0: farther than the storer + addr := swarm.NewAddress(b) + failing[addr.ByteString()] = true + peers = append(peers, addr) + } + + originBytes := chunk.Address().Clone().Bytes() + originBytes[0] ^= 0x80 + origin := swarm.NewAddress(originBytes) + + recorder := streamtest.New( + streamtest.WithPeerProtocols(map[string]p2p.ProtocolSpec{storer.String(): spec}), + streamtest.WithStreamError(func(addr swarm.Address, _, _, _ string) error { + if failing[addr.ByteString()] { + return errors.New("peer unreachable") + } + return nil + }), + streamtest.WithBaseAddr(origin), + ) + psOrigin, _ := createPushSyncNodeWithRadius(t, origin, defaultPrices, recorder, nil, nil, radius, mock.WithPeers(peers...)) + + return psOrigin, chunk, storer +} + +// TestOriginWaitsForInflightPush verifies that an origin whose push budget is +// spent still waits for a push in flight, which it pays for anyway. +func TestOriginWaitsForInflightPush(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + psOrigin, chunk, storer := slowStorerSetup(t, 10*time.Second) + + receipt, err := psOrigin.PushChunkToClosest(context.Background(), chunk) + if err != nil { + t.Fatalf("got %v, want the storer's receipt", err) + } + pub, err := crypto.Recover(receipt.Signature, receipt.Address.Bytes()) + if err != nil { + t.Fatal(err) + } + signedBy, err := crypto.NewOverlayAddress(*pub, 1, receipt.Nonce) + if err != nil { + t.Fatal(err) + } + if !signedBy.Equal(storer) { + t.Fatalf("receipt signed by %s, want the storer %s", signedBy, storer) + } + }) +} + +// TestOriginInflightWaitCancelled verifies that an origin waiting for a push in +// flight still returns as soon as its context is cancelled. +func TestOriginInflightWaitCancelled(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + const delay = 25 * time.Second + psOrigin, chunk, _ := slowStorerSetup(t, delay) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + time.AfterFunc(6*time.Second, cancel) + + start := time.Now() + if _, err := psOrigin.PushChunkToClosest(ctx, chunk); err == nil { + t.Fatal("got nil error, want the push to end") + } + if elapsed := time.Since(start); elapsed >= delay { + t.Fatalf("returned after %v, want before the storer answers at %v", elapsed, delay) + } + + // let the storer answer, so the push still in flight can end + time.Sleep(delay) + synctest.Wait() + }) +} From 50917bad7bf035f02e61ac7495c7dc80e63e5f83 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 15:01:03 +0200 Subject: [PATCH 07/11] fix(storer): do not count a cancelled requeued chunk as lost --- pkg/storer/export_test.go | 9 +++++++++ pkg/storer/netstore.go | 4 +++- pkg/storer/netstore_test.go | 4 ++++ 3 files changed, 16 insertions(+), 1 deletion(-) diff --git a/pkg/storer/export_test.go b/pkg/storer/export_test.go index 08c03aa3b1e..4abcb2ebac3 100644 --- a/pkg/storer/export_test.go +++ b/pkg/storer/export_test.go @@ -11,6 +11,7 @@ import ( "github.com/ethersphere/bee/v2/pkg/storer/internal/events" "github.com/ethersphere/bee/v2/pkg/storer/internal/reserve" "github.com/ethersphere/bee/v2/pkg/swarm" + dto "github.com/prometheus/client_model/go" ) func (db *DB) Reserve() *reserve.Reserve { @@ -71,3 +72,11 @@ func (r *Requeue) Drain() []swarm.Chunk { } return chunks } + +// DirectUploadRequeueFailed returns how many requeued direct upload chunks +// were counted as lost. +func (db *DB) DirectUploadRequeueFailed() float64 { + m := &dto.Metric{} + _ = db.metrics.DirectUploadRequeueFailed.Write(m) + return m.GetCounter().GetValue() +} diff --git a/pkg/storer/netstore.go b/pkg/storer/netstore.go index 34f0691615e..d19af333be4 100644 --- a/pkg/storer/netstore.go +++ b/pkg/storer/netstore.go @@ -97,7 +97,9 @@ func (db *DB) DirectUpload() PutterSession { // pushRequeued pushes a chunk that failed once for the second and last time. func (db *DB) pushRequeued(egCtx context.Context, r requeued) error { err := db.pushChunk(r.ctx, egCtx, r.ch) - if err != nil { + // a gone client, a failed upload or a closing node is not a chunk lost + // to the network + if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, ErrDBQuit) { db.metrics.DirectUploadRequeueFailed.Inc() db.logger.Debug("direct upload: chunk failed again", "chunk_address", r.ch.Address(), "error", err) } diff --git a/pkg/storer/netstore_test.go b/pkg/storer/netstore_test.go index af57231785e..23ea3c69574 100644 --- a/pkg/storer/netstore_test.go +++ b/pkg/storer/netstore_test.go @@ -695,6 +695,10 @@ func TestDirectUploadRequeue(t *testing.T) { if got := p.count(chunks[0].Address()); got != 1 { t.Fatalf("got %d pushes, want 1", got) } + // the client left; the chunk was not lost to the network + if got := lstore.DirectUploadRequeueFailed(); got != 0 { + t.Fatalf("got %v requeued chunks counted as lost, want 0", got) + } }) }) }) From 7a67bf71f378f27380fbadbe4f7f8a577fbbb362 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 15:01:41 +0200 Subject: [PATCH 08/11] fix(pusher): prune parked chunks at most once per park delay --- pkg/pusher/parked.go | 20 +++++++++++++------- pkg/pusher/parked_test.go | 27 +++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 7 deletions(-) diff --git a/pkg/pusher/parked.go b/pkg/pusher/parked.go index 85c8055ba81..245a56cfb05 100644 --- a/pkg/pusher/parked.go +++ b/pkg/pusher/parked.go @@ -26,10 +26,11 @@ const ( // parked tracks deferred chunks that failed their first round and wait in the // upload queue for their second and last one. type parked struct { - mu sync.Mutex - entries map[[64]byte]time.Time // identity address + batch ID -> until - delay time.Duration - ttl time.Duration + mu sync.Mutex + entries map[[64]byte]time.Time // identity address + batch ID -> until + delay time.Duration + ttl time.Duration + lastPrune time.Time } func newParked(delay, ttl time.Duration) *parked { @@ -55,10 +56,15 @@ func (p *parked) park(idAddress swarm.Address, batchID []byte, now time.Time) bo p.mu.Lock() defer p.mu.Unlock() - for k, until := range p.entries { - if now.After(until.Add(p.ttl)) { - delete(p.entries, k) + // walk the map at most once per delay: chunksWorker waits on the same + // lock, and a burst of failures parks many chunks at once + if now.Sub(p.lastPrune) >= p.delay { + for k, until := range p.entries { + if now.After(until.Add(p.ttl)) { + delete(p.entries, k) + } } + p.lastPrune = now } if _, ok := p.entries[key]; ok { diff --git a/pkg/pusher/parked_test.go b/pkg/pusher/parked_test.go index da7cdeef34a..ff9ec9ae35b 100644 --- a/pkg/pusher/parked_test.go +++ b/pkg/pusher/parked_test.go @@ -51,3 +51,30 @@ func TestParked(t *testing.T) { t.Fatal("park after the ttl: got false, want true") } } + +// TestParkedPrunesAtMostOncePerDelay tests that park scans for stale entries +// at most once per delay, so a burst of parks does not walk the whole map +// each time while chunksWorker waits on the same lock. +func TestParkedPrunesAtMostOncePerDelay(t *testing.T) { + t.Parallel() + + const delay = 5 * time.Minute + t0 := time.Date(2026, 10, 19, 0, 0, 0, 0, time.UTC) + batch := []byte("batch") + a, b := swarm.RandAddress(t), swarm.RandAddress(t) + + // no ttl: an entry is stale as soon as its delay has passed + p := pusher.NewParked(delay, 0) + + p.Park(a, batch, t0) // prunes + p.Park(b, batch, t0.Add(delay)) // prunes; a is not stale yet + + // a is stale now, but the last prune was less than a delay ago + if p.Park(a, batch, t0.Add(delay+time.Second)) { + t.Fatal("park within a delay of the last prune: got true, want false") + } + // a full delay after the last prune, stale entries are dropped + if !p.Park(a, batch, t0.Add(2*delay+time.Second)) { + t.Fatal("park a delay after the last prune: got false, want true") + } +} From 0cf553d3b90721d6fdfe11144880d98b501df5a0 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 22:14:52 +0200 Subject: [PATCH 09/11] fix(pusher): count a parked deferred chunk as pending, not as an error --- pkg/pusher/export_test.go | 12 ++++++ pkg/pusher/metrics.go | 2 +- pkg/pusher/pusher.go | 12 ++++-- pkg/pusher/pusher_test.go | 83 +++++++++++++++++++++++++++++++++++---- 4 files changed, 98 insertions(+), 11 deletions(-) diff --git a/pkg/pusher/export_test.go b/pkg/pusher/export_test.go index fe8c48f6295..97989439b7f 100644 --- a/pkg/pusher/export_test.go +++ b/pkg/pusher/export_test.go @@ -8,6 +8,8 @@ import ( "time" "github.com/ethersphere/bee/v2/pkg/swarm" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" ) const ParkDelay = parkDelay @@ -26,3 +28,13 @@ func (p *Parked) Waiting(idAddress swarm.Address, batchID []byte, now time.Time) } func (p *Parked) Forget(idAddress swarm.Address, batchID []byte) { p.p.forget(idAddress, batchID) } + +func counterValue(c prometheus.Counter) float64 { + m := &dto.Metric{} + _ = c.Write(m) + return m.GetCounter().GetValue() +} + +func (s *Service) TotalParked() float64 { return counterValue(s.metrics.TotalParked) } +func (s *Service) TotalCouldNotSync() float64 { return counterValue(s.metrics.TotalCouldNotSync) } +func (s *Service) TotalErrors() float64 { return counterValue(s.metrics.TotalErrors) } diff --git a/pkg/pusher/metrics.go b/pkg/pusher/metrics.go index 0eb3a2704f4..378a6f8ea08 100644 --- a/pkg/pusher/metrics.go +++ b/pkg/pusher/metrics.go @@ -40,7 +40,7 @@ func newMetrics() metrics { Namespace: m.Namespace, Subsystem: subsystem, Name: "total_could_not_sync", - Help: "Total chunks abandoned after exhausting retries with no valid receipt (shallow receipt or no peer in the correct neighbourhood).", + Help: "Deferred chunks given up on after their second round, and direct upload rounds that ended with no valid receipt (a direct chunk is lost only when localstore_direct_upload_requeue_failed counts it).", }), TotalParked: prometheus.NewCounter(prometheus.CounterOpts{ Namespace: m.Namespace, diff --git a/pkg/pusher/pusher.go b/pkg/pusher/pusher.go index c68c82b05e3..bffd09f2a0b 100644 --- a/pkg/pusher/pusher.go +++ b/pkg/pusher/pusher.go @@ -336,17 +336,23 @@ func (s *Service) parkOrGiveUp(ctx context.Context, logger log.Logger, op *Op, e if s.parked.park(op.identityAddress, batchID, time.Now()) { s.metrics.TotalParked.Inc() logger.Debug("chunk not placed, will retry later", "chunk_address", op.Chunk.Address(), "retry_in", parkDelay, "error", err) - return false, err + // the chunk waits for its second round, which is not an error yet + if op.Span != nil { + op.Span.SetAttributes(attribute.Bool("swarm.chunk.parked", true)) + } + return false, nil } - s.parked.forget(op.identityAddress, batchID) + logger.Debug("chunk not placed after its second round, giving up", "chunk_address", op.Chunk.Address(), "error", err) // report CouldNotSync so the chunk isn't marked delivered s.metrics.TotalCouldNotSync.Inc() if err := s.storer.Report(ctx, op.Chunk, storage.ChunkCouldNotSync); err != nil { logger.Error(err, "pusher: failed to report sync status") + // keep the entry: the retry must give up again, not park once more return true, err } - return false, nil + s.parked.forget(op.identityAddress, batchID) + return false, err } func (s *Service) pushDirect(ctx context.Context, logger log.Logger, op *Op) error { diff --git a/pkg/pusher/pusher_test.go b/pkg/pusher/pusher_test.go index 8e51f299446..79ad02acdd8 100644 --- a/pkg/pusher/pusher_test.go +++ b/pkg/pusher/pusher_test.go @@ -383,13 +383,15 @@ func TestShallowReceiptExhausted(t *testing.T) { calls.Add(1) return &pushsync.Receipt{Address: chunk.Address()}, pushsync.ErrShallowReceipt }) - storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) + svc, storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) offer(storer, chunk) if got := int(calls.Load()); got != defaultRetryCount { t.Fatalf("round 1: got %d push attempts, want %d", got, defaultRetryCount) } expectNoReport(t, storer) + // every shallow receipt but the last is an error; parking is not + expectCounters(t, svc, 1, 0, float64(defaultRetryCount-1)) // offered again while parked: skipped offer(storer, chunk) @@ -403,6 +405,8 @@ func TestShallowReceiptExhausted(t *testing.T) { t.Fatalf("round 2: got %d push attempts, want %d", got, 2*defaultRetryCount) } expectReport(t, storer, storage.ChunkCouldNotSync) + // giving up is + expectCounters(t, svc, 1, 1, float64(2*(defaultRetryCount-1)+1)) }) }) @@ -456,7 +460,7 @@ func TestOutOfDepthStoring(t *testing.T) { calls.Add(1) return nil, pushsync.ErrOutOfDepthStoring }) - storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) + svc, storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) // the queue offers the chunk again and again while it is pending for range 3 { @@ -466,6 +470,7 @@ func TestOutOfDepthStoring(t *testing.T) { t.Fatalf("got %d push attempts before the park delay, want 1", got) } expectNoReport(t, storer) + expectCounters(t, svc, 1, 0, 0) time.Sleep(pusher.ParkDelay) offer(storer, chunk) @@ -473,6 +478,7 @@ func TestOutOfDepthStoring(t *testing.T) { t.Fatalf("got %d push attempts after the park delay, want 2", got) } expectReport(t, storer, storage.ChunkCouldNotSync) + expectCounters(t, svc, 1, 1, 1) }) }) @@ -960,17 +966,37 @@ func TestDirectBatchGone(t *testing.T) { // newParkingPusher starts a pusher over a storer whose push queue the test // drives with offer. It must be called inside a synctest bubble. -func newParkingPusher(t *testing.T, pushSyncService pushsync.PushSyncer, batchExist postage.BatchExist) (*reportingStorer, swarm.Chunk) { +func newParkingPusher(t *testing.T, pushSyncService pushsync.PushSyncer, batchExist postage.BatchExist) (*pusher.Service, *reportingStorer, swarm.Chunk) { t.Helper() storer := &reportingStorer{ mockStorer: &mockStorer{chunks: make(chan swarm.Chunk)}, reported: make(chan storage.ChunkState, 8), } + return newParkingPusherWith(t, storer, pushSyncService, batchExist), storer, testingc.GenerateTestRandomChunk() +} + +func newParkingPusherWith(t *testing.T, storer pusher.Storer, pushSyncService pushsync.PushSyncer, batchExist postage.BatchExist) *pusher.Service { + t.Helper() + pusherSvc := pusher.New(1, storer, pushSyncService, batchExist, log.Noop, stabilmock.NewSubscriber(true), defaultRetryCount) t.Cleanup(func() { _ = pusherSvc.Close() }) + return pusherSvc +} - return storer, testingc.GenerateTestRandomChunk() +// expectCounters checks the pusher counters that tell a parked chunk from a +// lost one. +func expectCounters(t *testing.T, svc *pusher.Service, parked, couldNotSync, errs float64) { + t.Helper() + if got := svc.TotalParked(); got != parked { + t.Fatalf("total_parked: got %v, want %v", got, parked) + } + if got := svc.TotalCouldNotSync(); got != couldNotSync { + t.Fatalf("total_could_not_sync: got %v, want %v", got, couldNotSync) + } + if got := svc.TotalErrors(); got != errs { + t.Fatalf("total_errors: got %v, want %v", got, errs) + } } // offer hands the chunk to the pusher as the upload store iteration does and @@ -1014,7 +1040,7 @@ func TestParkedChunkSynced(t *testing.T) { } return nil, pushsync.ErrOutOfDepthStoring }) - storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) + _, storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) offer(storer, chunk) expectNoReport(t, storer) @@ -1042,7 +1068,7 @@ func TestParkedChunkBatchGone(t *testing.T) { batchExist := batchstoremock.New(batchstoremock.WithExistsFunc(func(b []byte) (bool, error) { return !batchGone.Load(), nil })) - storer, chunk := newParkingPusher(t, pushSyncService, batchExist) + _, storer, chunk := newParkingPusher(t, pushSyncService, batchExist) offer(storer, chunk) expectNoReport(t, storer) @@ -1072,7 +1098,7 @@ func TestDeferredOutOfDepthResetsAttempts(t *testing.T) { } return &pushsync.Receipt{Address: chunk.Address()}, pushsync.ErrShallowReceipt }) - storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) + _, storer, chunk := newParkingPusher(t, pushSyncService, defaultMockBatchStore) offer(storer, chunk) if got := calls.Load(); got != 2 { @@ -1088,3 +1114,46 @@ func TestDeferredOutOfDepthResetsAttempts(t *testing.T) { expectReport(t, storer, storage.ChunkCouldNotSync) }) } + +// failingReportStorer fails the first ChunkCouldNotSync report. +type failingReportStorer struct { + *reportingStorer + failed atomic.Bool +} + +func (f *failingReportStorer) Report(ctx context.Context, chunk swarm.Chunk, state storage.ChunkState) error { + if state == storage.ChunkCouldNotSync && f.failed.CompareAndSwap(false, true) { + return errors.New("report failed") + } + return f.reportingStorer.Report(ctx, chunk, state) +} + +// TestParkedGiveUpReportFails tests that a chunk whose give-up report fails is +// given up on again when it is retried, not parked for a third round. +func TestParkedGiveUpReportFails(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + pushSyncService := pushsyncmock.New(func(ctx context.Context, chunk swarm.Chunk) (*pushsync.Receipt, error) { + return nil, pushsync.ErrOutOfDepthStoring + }) + storer := &failingReportStorer{reportingStorer: &reportingStorer{ + mockStorer: &mockStorer{chunks: make(chan swarm.Chunk)}, + reported: make(chan storage.ChunkState, 8), + }} + svc := newParkingPusherWith(t, storer, pushSyncService, defaultMockBatchStore) + chunk := testingc.GenerateTestRandomChunk() + + storer.chunks <- chunk + synctest.Wait() + + time.Sleep(pusher.ParkDelay) + storer.chunks <- chunk + synctest.Wait() + + expectReport(t, storer.reportingStorer, storage.ChunkCouldNotSync) + if got := svc.TotalParked(); got != 1 { + t.Fatalf("total_parked: got %v, want 1", got) + } + }) +} From fd6f6ca74de1126d80ecc2eabfb7a1cec3b85fbc Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 19 Oct 2026 22:16:24 +0200 Subject: [PATCH 10/11] test(storer,pushsync): tighten upload retry tests and use chunk_address in logs --- pkg/pushsync/pushsync_test.go | 10 ++++++---- pkg/storer/netstore.go | 4 ++-- pkg/storer/netstore_test.go | 13 ++++++++++++- 3 files changed, 20 insertions(+), 7 deletions(-) diff --git a/pkg/pushsync/pushsync_test.go b/pkg/pushsync/pushsync_test.go index 8cd3c779501..69b985375e5 100644 --- a/pkg/pushsync/pushsync_test.go +++ b/pkg/pushsync/pushsync_test.go @@ -1514,11 +1514,13 @@ func TestOriginInflightWaitCancelled(t *testing.T) { time.AfterFunc(6*time.Second, cancel) start := time.Now() - if _, err := psOrigin.PushChunkToClosest(ctx, chunk); err == nil { - t.Fatal("got nil error, want the push to end") + if _, err := psOrigin.PushChunkToClosest(ctx, chunk); !errors.Is(err, pushsync.ErrNoPush) { + t.Fatalf("got %v, want %v", err, pushsync.ErrNoPush) } - if elapsed := time.Since(start); elapsed >= delay { - t.Fatalf("returned after %v, want before the storer answers at %v", elapsed, delay) + // the origin was still waiting when the context was cancelled, and + // it did not wait for the storer + if elapsed := time.Since(start); elapsed < 6*time.Second || elapsed >= delay { + t.Fatalf("returned after %v, want at the cancel (6s), before the storer answers at %v", elapsed, delay) } // let the storer answer, so the push still in flight can end diff --git a/pkg/storer/netstore.go b/pkg/storer/netstore.go index d19af333be4..6ff436c26b9 100644 --- a/pkg/storer/netstore.go +++ b/pkg/storer/netstore.go @@ -145,9 +145,9 @@ func (db *DB) pushChunk(ctx, egCtx context.Context, ch swarm.Chunk) (err error) return ErrDBQuit case err := <-op.Err: if errors.Is(err, pushsync.ErrShallowReceipt) { - logger.Debug("direct upload: shallow receipt received, retrying", "chunk", ch.Address()) + logger.Debug("direct upload: shallow receipt received, retrying", "chunk_address", ch.Address()) } else if errors.Is(err, topology.ErrNotFound) { - logger.Debug("direct upload: no peers available, retrying", "chunk", ch.Address()) + logger.Debug("direct upload: no peers available, retrying", "chunk_address", ch.Address()) // wait for peers instead of resubmitting in a tight loop select { case <-time.After(directUploadRetryDelay): diff --git a/pkg/storer/netstore_test.go b/pkg/storer/netstore_test.go index 23ea3c69574..44865827c8f 100644 --- a/pkg/storer/netstore_test.go +++ b/pkg/storer/netstore_test.go @@ -616,13 +616,24 @@ func TestDirectUploadRequeue(t *testing.T) { t.Run("full queue oldest fails again", func(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { - chunks := chunktesting.GenerateTestRandomChunks(storer.DirectUploadRequeueLimit + 1) + chunks := chunktesting.GenerateTestRandomChunks(storer.DirectUploadRequeueLimit + 2) + next := chunks[len(chunks)-1] + chunks = chunks[:len(chunks)-1] withScriptedPusher(t, failFirst(chunks, chunks[:1]), func(lstore *storer.DB, p *scriptedPusher) { session := lstore.DirectUpload() for _, ch := range chunks { put(t, session, ch) } + // the upload has failed: the next chunk is not pushed + if err := session.Put(context.Background(), next); !errors.Is(err, pusher.ErrCouldNotSync) { + t.Fatalf("session.Put(next): got %v, want %v", err, pusher.ErrCouldNotSync) + } + synctest.Wait() + if got := p.count(next.Address()); got != 0 { + t.Fatalf("next chunk: got %d pushes, want 0", got) + } + if err := session.Done(swarm.ZeroAddress); !errors.Is(err, pusher.ErrCouldNotSync) { t.Fatalf("session.Done(): got %v, want %v", err, pusher.ErrCouldNotSync) } From 3b4bf356b8b72e987bb64f7d0ce41adab7d41293 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Fri, 23 Oct 2026 13:16:21 +0200 Subject: [PATCH 11/11] fix(storer): wait out an outage before retrying a requeued chunk --- pkg/storer/export_test.go | 16 +++++++++----- pkg/storer/netstore.go | 17 ++++++++++++-- pkg/storer/netstore_test.go | 14 ++++++++++++ pkg/storer/requeue.go | 44 +++++++++++++++++++++++-------------- pkg/storer/requeue_test.go | 42 +++++++++++++++++++++++++++++------ 5 files changed, 102 insertions(+), 31 deletions(-) diff --git a/pkg/storer/export_test.go b/pkg/storer/export_test.go index 4abcb2ebac3..119317341ed 100644 --- a/pkg/storer/export_test.go +++ b/pkg/storer/export_test.go @@ -6,6 +6,7 @@ package storer import ( "context" + "time" "github.com/ethersphere/bee/v2/pkg/bmt" "github.com/ethersphere/bee/v2/pkg/storer/internal/events" @@ -52,16 +53,21 @@ func TransformedAddress(hasher bmt.Hasher, ch swarm.Chunk, chType swarm.ChunkTyp return transformedAddress(hasher, ch.Address(), ch.Data(), chType) } -const DirectUploadRequeueLimit = directUploadRequeueLimit +const ( + DirectUploadRequeueLimit = directUploadRequeueLimit + DirectUploadRequeueMinAge = directUploadRequeueMinAge +) // Requeue exposes requeue to tests with chunks only. type Requeue struct{ q *requeue } -func NewRequeue(limit int) *Requeue { return &Requeue{q: newRequeue(limit)} } +func NewRequeue(limit int, minAge time.Duration) *Requeue { + return &Requeue{q: newRequeue(limit, minAge)} +} -func (r *Requeue) Add(ch swarm.Chunk) (swarm.Chunk, bool) { - oldest, full := r.q.add(requeued{ctx: context.Background(), ch: ch}) - return oldest.ch, full +func (r *Requeue) Add(ch swarm.Chunk, now time.Time) (swarm.Chunk, time.Duration, bool) { + oldest, wait, full := r.q.add(requeued{ctx: context.Background(), ch: ch}, now) + return oldest.ch, wait, full } func (r *Requeue) Drain() []swarm.Chunk { diff --git a/pkg/storer/netstore.go b/pkg/storer/netstore.go index 6ff436c26b9..8e58a7b27b1 100644 --- a/pkg/storer/netstore.go +++ b/pkg/storer/netstore.go @@ -32,7 +32,7 @@ func (db *DB) DirectUpload() PutterSession { // chunks that failed once wait here for one more push at the end of the // upload - q := newRequeue(directUploadRequeueLimit) + q := newRequeue(directUploadRequeueLimit, directUploadRequeueMinAge) return &putterSession{ Putter: putterWithMetrics{ @@ -57,10 +57,23 @@ func (db *DB) DirectUpload() PutterSession { db.metrics.DirectUploadRequeued.Inc() db.logger.Debug("direct upload: chunk failed, will push it again at the end of the upload", "chunk_address", ch.Address()) - oldest, full := q.add(requeued{ctx: ctx, ch: ch}) + oldest, wait, full := q.add(requeued{ctx: ctx, ch: ch}, time.Now()) if !full { return nil } + // a full queue means an outage: let it pass before the + // second push, holding the upload back meanwhile + if wait > 0 { + select { + case <-time.After(wait): + case <-ctx.Done(): + return ctx.Err() + case <-egCtx.Done(): + return context.Cause(egCtx) + case <-db.quit: + return ErrDBQuit + } + } return db.pushRequeued(egCtx, oldest) })) return nil diff --git a/pkg/storer/netstore_test.go b/pkg/storer/netstore_test.go index 44865827c8f..2431c178365 100644 --- a/pkg/storer/netstore_test.go +++ b/pkg/storer/netstore_test.go @@ -598,6 +598,12 @@ func TestDirectUploadRequeue(t *testing.T) { put(t, session, ch) } + // the oldest chunk waits out the minimum age first + if got := p.count(chunks[0].Address()); got != 1 { + t.Fatalf("oldest chunk: got %d pushes while the queue is full, want 1", got) + } + time.Sleep(storer.DirectUploadRequeueMinAge) + synctest.Wait() if got := p.count(chunks[0].Address()); got != 2 { t.Fatalf("oldest chunk: got %d pushes before done, want 2", got) } @@ -625,7 +631,15 @@ func TestDirectUploadRequeue(t *testing.T) { put(t, session, ch) } + // the upload is alive while the oldest chunk waits + if err := session.Put(context.Background(), next); err != nil { + t.Fatalf("session.Put(next) while waiting: unexpected error: %v", err) + } + time.Sleep(storer.DirectUploadRequeueMinAge) + synctest.Wait() + // the upload has failed: the next chunk is not pushed + next = chunktesting.GenerateTestRandomChunk() if err := session.Put(context.Background(), next); !errors.Is(err, pusher.ErrCouldNotSync) { t.Fatalf("session.Put(next): got %v, want %v", err, pusher.ErrCouldNotSync) } diff --git a/pkg/storer/requeue.go b/pkg/storer/requeue.go index 9af1dec79ef..1a0e389afe2 100644 --- a/pkg/storer/requeue.go +++ b/pkg/storer/requeue.go @@ -7,46 +7,56 @@ package storer import ( "context" "sync" + "time" "github.com/ethersphere/bee/v2/pkg/swarm" ) -// directUploadRequeueLimit is how many chunks of one direct upload may wait -// for their second push at the same time. Many failures in a short time point -// to a problem that a later push will not fix, so the upload then fails fast. -const directUploadRequeueLimit = 16 +const ( + // directUploadRequeueLimit is how many chunks of one direct upload may + // wait for their second push. When it is reached, the oldest is pushed + // again, and the upload fails if that push fails too. + directUploadRequeueLimit = 16 + + // directUploadRequeueMinAge is how long a chunk waits before its second + // push when the queue is full, so that a short outage can pass first. + directUploadRequeueMinAge = 30 * time.Second +) // requeued is a chunk of a direct upload waiting for its second push. type requeued struct { - ctx context.Context // the Put context, still alive at Done - ch swarm.Chunk + ctx context.Context // the Put context, still alive at Done + ch swarm.Chunk + since time.Time // when the chunk was queued } // requeue holds the chunks of a direct upload that failed once. They are // pushed again at the end of the upload. type requeue struct { - mu sync.Mutex - limit int - items []requeued + mu sync.Mutex + limit int + minAge time.Duration + items []requeued } -func newRequeue(limit int) *requeue { - return &requeue{limit: limit} +func newRequeue(limit int, minAge time.Duration) *requeue { + return &requeue{limit: limit, minAge: minAge} } -// add appends r. When the queue is full, it removes the oldest entry in the -// same step and returns it; the caller must push that one now. -func (q *requeue) add(r requeued) (requeued, bool) { +// add appends r. When the queue is full, it removes and returns the oldest +// entry and how long it still has to wait; the caller must push it. +func (q *requeue) add(r requeued, now time.Time) (oldest requeued, wait time.Duration, full bool) { q.mu.Lock() defer q.mu.Unlock() + r.since = now if len(q.items) < q.limit { q.items = append(q.items, r) - return requeued{}, false + return requeued{}, 0, false } - oldest := q.items[0] + oldest = q.items[0] q.items = append(q.items[1:], r) - return oldest, true + return oldest, max(0, q.minAge-now.Sub(oldest.since)), true } // drain returns the entries, oldest first, and empties the queue. diff --git a/pkg/storer/requeue_test.go b/pkg/storer/requeue_test.go index 60ab94b795a..5c53e62b7ed 100644 --- a/pkg/storer/requeue_test.go +++ b/pkg/storer/requeue_test.go @@ -7,6 +7,7 @@ package storer_test import ( "sync" "testing" + "time" chunktesting "github.com/ethersphere/bee/v2/pkg/storage/testing" "github.com/ethersphere/bee/v2/pkg/storer" @@ -17,18 +18,22 @@ func TestRequeueAdd(t *testing.T) { t.Parallel() chunks := chunktesting.GenerateTestRandomChunks(3) - q := storer.NewRequeue(2) + q := storer.NewRequeue(2, 0) + now := time.Now() - if _, full := q.Add(chunks[0]); full { + if _, _, full := q.Add(chunks[0], now); full { t.Fatal("add 1: queue reported full") } - if _, full := q.Add(chunks[1]); full { + if _, _, full := q.Add(chunks[1], now); full { t.Fatal("add 2: queue reported full") } - oldest, full := q.Add(chunks[2]) + oldest, wait, full := q.Add(chunks[2], now) if !full { t.Fatal("add 3: queue not reported full") } + if wait != 0 { + t.Fatalf("add 3: got wait %v, want 0", wait) + } if !oldest.Address().Equal(chunks[0].Address()) { t.Fatalf("add 3: got oldest %s, want %s", oldest.Address(), chunks[0].Address()) } @@ -42,13 +47,36 @@ func TestRequeueAdd(t *testing.T) { } } +// TestRequeueMinAge verifies the wait returned for the oldest entry of a +// full queue. +func TestRequeueMinAge(t *testing.T) { + t.Parallel() + + const minAge = 30 * time.Second + chunks := chunktesting.GenerateTestRandomChunks(3) + q := storer.NewRequeue(1, minAge) + start := time.Now() + + q.Add(chunks[0], start) + + // 10 s old: 20 s to go + if _, wait, full := q.Add(chunks[1], start.Add(10*time.Second)); !full || wait != 20*time.Second { + t.Fatalf("add 2: got full=%v wait=%v, want true 20s", full, wait) + } + // old enough: no wait + if _, wait, full := q.Add(chunks[2], start.Add(time.Minute)); !full || wait != 0 { + t.Fatalf("add 3: got full=%v wait=%v, want true 0", full, wait) + } +} + func TestRequeueConcurrentAdd(t *testing.T) { t.Parallel() limit := storer.DirectUploadRequeueLimit - q := storer.NewRequeue(limit) + q := storer.NewRequeue(limit, 0) + now := time.Now() for _, ch := range chunktesting.GenerateTestRandomChunks(limit) { - q.Add(ch) + q.Add(ch, now) } added := chunktesting.GenerateTestRandomChunks(64) @@ -58,7 +86,7 @@ func TestRequeueConcurrentAdd(t *testing.T) { var wg sync.WaitGroup for i, ch := range added { wg.Go(func() { - returned[i], fulls[i] = q.Add(ch) + returned[i], _, fulls[i] = q.Add(ch, now) }) } wg.Wait()