diff --git a/pkg/pusher/export_test.go b/pkg/pusher/export_test.go
new file mode 100644
index 00000000000..97989439b7f
--- /dev/null
+++ b/pkg/pusher/export_test.go
@@ -0,0 +1,40 @@
+// 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"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+)
+
+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) }
+
+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 d0000cce13b..378a6f8ea08 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
@@ -39,7 +40,13 @@ 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,
+			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,
diff --git a/pkg/pusher/parked.go b/pkg/pusher/parked.go
new file mode 100644
index 00000000000..245a56cfb05
--- /dev/null
+++ b/pkg/pusher/parked.go
@@ -0,0 +1,96 @@
+// 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
+	lastPrune time.Time
+}
+
+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()
+
+	// 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 {
+		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..ff9ec9ae35b
--- /dev/null
+++ b/pkg/pusher/parked_test.go
@@ -0,0 +1,80 @@
+// 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")
+	}
+}
+
+// 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")
+	}
+}
diff --git a/pkg/pusher/pusher.go b/pkg/pusher/pusher.go
index 7b1b04389d1..bffd09f2a0b 100644
--- a/pkg/pusher/pusher.go
+++ b/pkg/pusher/pusher.go
@@ -12,6 +12,7 @@ import (
 	"context"
 	"encoding/hex"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -60,6 +61,7 @@ type Service struct {
 	chunksWorkerQuitC chan struct{}
 	inflight          *inflight
 	attempts          *attempts
+	parked            *parked
 	smuggler          chan OpChan
 }
 
@@ -93,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)
@@ -237,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 {
@@ -275,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))
 	}
 
@@ -294,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")
@@ -311,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
@@ -319,6 +325,36 @@ 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)
+		// 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
+	}
+
+	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
+	}
+	s.parked.forget(op.identityAddress, batchID)
+	return false, err
+}
+
 func (s *Service) pushDirect(ctx context.Context, logger log.Logger, op *Op) error {
 	loggerV1 := logger.V(1).Build()
 
@@ -341,6 +377,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
 	}
 
@@ -370,6 +410,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..79ad02acdd8 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,41 @@ 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)
-
-		before := callCount.Load()
-		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 &pushsync.Receipt{Address: chunk.Address()}, pushsync.ErrShallowReceipt
+			})
+			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)
+			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)
+			// giving up is
+			expectCounters(t, svc, 1, 1, float64(2*(defaultRetryCount-1)+1))
+		})
 	})
 
 	t.Run("direct", func(t *testing.T) {
@@ -430,40 +444,52 @@ 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
+			})
+			svc, 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)
+			expectCounters(t, svc, 1, 0, 0)
 
-		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)
+			expectCounters(t, svc, 1, 1, 1)
+		})
 	})
 
+	// 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 +499,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)
 		}
@@ -863,3 +888,272 @@ 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)
+	}
+}
+
+// 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)
+	}
+}
+
+// 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) (*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
+}
+
+// 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
+// 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)
+	})
+}
+
+// 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)
+		}
+	})
+}
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..69b985375e5 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,126 @@ 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); !errors.Is(err, pushsync.ErrNoPush) {
+			t.Fatalf("got %v, want %v", err, pushsync.ErrNoPush)
+		}
+		// 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
+		time.Sleep(delay)
+		synctest.Wait()
+	})
+}
diff --git a/pkg/storer/export_test.go b/pkg/storer/export_test.go
index d9b779ba97f..119317341ed 100644
--- a/pkg/storer/export_test.go
+++ b/pkg/storer/export_test.go
@@ -5,10 +5,14 @@
 package storer
 
 import (
+	"context"
+	"time"
+
 	"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"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
+	dto "github.com/prometheus/client_model/go"
 )
 
 func (db *DB) Reserve() *reserve.Reserve {
@@ -48,3 +52,37 @@ 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
+	DirectUploadRequeueMinAge = directUploadRequeueMinAge
+)
+
+// Requeue exposes requeue to tests with chunks only.
+type Requeue struct{ q *requeue }
+
+func NewRequeue(limit int, minAge time.Duration) *Requeue {
+	return &Requeue{q: newRequeue(limit, minAge)}
+}
+
+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 {
+	items := r.q.drain()
+	chunks := make([]swarm.Chunk, len(items))
+	for i, it := range items {
+		chunks[i] = it.ch
+	}
+	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/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 bf7d3fc5541..8e58a7b27b1 100644
--- a/pkg/storer/netstore.go
+++ b/pkg/storer/netstore.go
@@ -30,71 +30,155 @@ 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, directUploadRequeueMinAge)
+
 	return &putterSession{
 		Putter: putterWithMetrics{
 			storage.PutterFunc(func(ctx context.Context, ch swarm.Chunk) error {
-				db.directUploadLimiter <- struct{}{}
-				eg.Go(safe.RunFunc(db.logger, "storer-netstore-direct-upload", func() (err error) {
+				// 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() 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}
+					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 egCtx.Err()
+							return context.Cause(egCtx)
 						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
-								}
-							}
 						}
 					}
+					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)
+	// 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)
+	}
+	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_address", ch.Address())
+				} else if errors.Is(err, topology.ErrNotFound) {
+					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):
+					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 302d0c055f6..2431c178365 100644
--- a/pkg/storer/netstore_test.go
+++ b/pkg/storer/netstore_test.go
@@ -9,7 +9,9 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"sync"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/ethersphere/bee/v2/pkg/pusher"
@@ -388,3 +390,341 @@ 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)
+		}
+	})
+}
+
+// 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)
+				}
+
+				// 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)
+				}
+				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 + 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 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)
+				}
+				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)
+				}
+				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)
+				}
+				// 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)
+				}
+			})
+		})
+	})
+}
diff --git a/pkg/storer/requeue.go b/pkg/storer/requeue.go
new file mode 100644
index 00000000000..1a0e389afe2
--- /dev/null
+++ b/pkg/storer/requeue.go
@@ -0,0 +1,70 @@
+// 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"
+	"time"
+
+	"github.com/ethersphere/bee/v2/pkg/swarm"
+)
+
+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
+	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
+	minAge time.Duration
+	items  []requeued
+}
+
+func newRequeue(limit int, minAge time.Duration) *requeue {
+	return &requeue{limit: limit, minAge: minAge}
+}
+
+// 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{}, 0, false
+	}
+	oldest = q.items[0]
+	q.items = append(q.items[1:], r)
+	return oldest, max(0, q.minAge-now.Sub(oldest.since)), 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..5c53e62b7ed
--- /dev/null
+++ b/pkg/storer/requeue_test.go
@@ -0,0 +1,116 @@
+// 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"
+	"time"
+
+	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, 0)
+	now := time.Now()
+
+	if _, _, full := q.Add(chunks[0], now); full {
+		t.Fatal("add 1: queue reported full")
+	}
+	if _, _, full := q.Add(chunks[1], now); full {
+		t.Fatal("add 2: queue reported full")
+	}
+	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())
+	}
+
+	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))
+	}
+}
+
+// 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, 0)
+	now := time.Now()
+	for _, ch := range chunktesting.GenerateTestRandomChunks(limit) {
+		q.Add(ch, now)
+	}
+
+	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, now)
+		})
+	}
+	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
+}
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
 			}
 		}
 	})