feat(storer): improve SOC/CAC stamp collision convergence for pullsync

Author: sbackend123

pkg/puller/puller.go

@@ -396,12 +396,35 @@ func (p *Puller) syncPeerBin(parentCtx context.Context, peer *syncPeer, bin uint
 
 			// pulled at least one chunk
 			if top >= start {
+				if err != nil {
+					p.logger.Debug("syncWorker advancing interval despite sync errors",
+						"error", err,
+						"peer_address", address,
+						"bin", bin,
+						"cursor", cursor,
+						"start", start,
+						"topmost", top,
+						"count", count,
+						"historical", isHistorical,
+					)
+				}
 				if err := p.addPeerInterval(address, bin, start, top); err != nil {
 					p.metrics.SyncWorkerErrCounter.Inc()
 					p.logger.Error(err, "syncWorker could not persist interval for peer, quitting", "peer_address", address)
 					return
 				}
 				start = top + 1
+			} else if err != nil {
+				p.logger.Debug("syncWorker not advancing interval after sync error",
+					"error", err,
+					"peer_address", address,
+					"bin", bin,
+					"cursor", cursor,
+					"start", start,
+					"topmost", top,
+					"count", count,
+					"historical", isHistorical,
+				)
 			}
 		}
 	}

pkg/pullsync/pullsync.go

@@ -8,6 +8,7 @@ package pullsync
 
 import (
 	"context"
+	"encoding/binary"
 	"encoding/hex"
 	"errors"
 	"fmt"
@@ -303,7 +304,32 @@ func (s *Syncer) Sync(ctx context.Context, peer swarm.Address, bin uint8, start
 				ctr++
 				s.metrics.Wanted.Inc()
 				bv.Set(i)
+				s.logger.Debug("pullsync want chunk",
+					"peer_address", peer,
+					"bin", bin,
+					"offer_idx", i,
+					"chunk_address", a,
+					"sum", hex.EncodeToString(sum),
+					"start", start,
+					"offer_topmost", topmost,
+				)
+			} else {
+				s.logger.Debug("pullsync skip offer, already have sum",
+					"peer_address", peer,
+					"bin", bin,
+					"offer_idx", i,
+					"chunk_address", a,
+					"sum", hex.EncodeToString(sum),
+					"start", start,
+					"offer_topmost", topmost,
+				)
 			}
+		} else {
+			s.logger.Debug("pullsync skip offer, outside storage radius",
+				"peer_address", peer,
+				"bin", bin,
+				"chunk_address", a,
+			)
 		}
 	}
 
@@ -313,6 +339,7 @@ func (s *Syncer) Sync(ctx context.Context, peer swarm.Address, bin uint8, start
 	}
 
 	chunksToPut := make([]swarm.Chunk, 0, ctr)
+	wanted := ctr
 
 	var chunkErr error
 	for ; ctr > 0; ctr-- {
@@ -345,17 +372,53 @@ func (s *Syncer) Sync(ctx context.Context, peer swarm.Address, bin uint8, start
 		}
 
 		wantChunkID := addr.ByteString() + string(sum)
+		stampHash, hashErr := stamp.Hash()
+		stampHashHex := ""
+		if hashErr == nil {
+			stampHashHex = hex.EncodeToString(stampHash)
+		}
 		if _, ok := wantChunks[wantChunkID]; !ok {
-			s.logger.Debug("want chunks", "error", ErrUnsolicitedChunk, "peer_address", peer, "chunk_address", addr)
+			s.logger.Debug("pullsync unsolicited delivery",
+				"error", ErrUnsolicitedChunk,
+				"peer_address", peer,
+				"chunk_address", addr,
+				"batch_id", hex.EncodeToString(stamp.BatchID()),
+				"stamp_hash", stampHashHex,
+				"stamp_index", hex.EncodeToString(stamp.Index()),
+				"stamp_timestamp", binary.BigEndian.Uint64(stamp.Timestamp()),
+				"recomputed_sum", hex.EncodeToString(sum),
+				"bin", bin,
+				"offer_topmost", topmost,
+			)
 			chunkErr = errors.Join(chunkErr, ErrUnsolicitedChunk)
 			continue
 		}
 
 		delete(wantChunks, wantChunkID)
 
+		s.logger.Debug("pullsync delivery accepted for want",
+			"peer_address", peer,
+			"chunk_address", addr,
+			"batch_id", hex.EncodeToString(stamp.BatchID()),
+			"stamp_hash", stampHashHex,
+			"stamp_index", hex.EncodeToString(stamp.Index()),
+			"stamp_timestamp", binary.BigEndian.Uint64(stamp.Timestamp()),
+			"recomputed_sum", hex.EncodeToString(sum),
+			"bin", bin,
+			"offer_topmost", topmost,
+		)
+
 		chunk, err := s.validStamp(newChunk.WithStamp(stamp))
 		if err != nil {
-			s.logger.Debug("unverified stamp", "error", err, "peer_address", peer, "chunk_address", newChunk)
+			s.logger.Debug("unverified stamp",
+				"error", err,
+				"peer_address", peer,
+				"chunk_address", addr,
+				"batch_id", hex.EncodeToString(stamp.BatchID()),
+				"stamp_timestamp", binary.BigEndian.Uint64(stamp.Timestamp()),
+				"bin", bin,
+				"offer_topmost", topmost,
+			)
 			chunkErr = errors.Join(chunkErr, err)
 			continue
 		}
@@ -390,24 +453,66 @@ func (s *Syncer) Sync(ctx context.Context, peer swarm.Address, bin uint8, start
 				// in case of these errors, no new items are added to the storage, so it
 				// is safe to continue with the next chunk
 				if errors.Is(err, storage.ErrOverwriteNewerChunk) {
-					s.logger.Debug("overwrite newer chunk", "error", err, "peer_address", peer, "chunk", c)
+					s.logger.Debug("overwrite newer chunk",
+						"error", err,
+						"peer_address", peer,
+						"chunk_address", c.Address(),
+						"batch_id", hex.EncodeToString(c.Stamp().BatchID()),
+						"stamp_timestamp", binary.BigEndian.Uint64(c.Stamp().Timestamp()),
+						"bin", bin,
+					)
 					chunkErr = errors.Join(chunkErr, err)
 					continue
 				}
 				// the chunk diverged from the one already stored and lost the
 				// tie-break. The neighborhood converges on the stored chunk, so
 				// this is an expected outcome rather than a sync error.
 				if errors.Is(err, storage.ErrDivergentChunkRejected) {
-					s.logger.Debug("divergent chunk rejected", "error", err, "peer_address", peer, "chunk", c)
+					s.logger.Debug("divergent chunk rejected",
+						"error", err,
+						"peer_address", peer,
+						"chunk_address", c.Address(),
+						"batch_id", hex.EncodeToString(c.Stamp().BatchID()),
+						"stamp_timestamp", binary.BigEndian.Uint64(c.Stamp().Timestamp()),
+						"bin", bin,
+					)
 					s.metrics.DivergentRejected.Inc()
 					continue
 				}
+				s.logger.Debug("pullsync reserve put failed",
+					"error", err,
+					"peer_address", peer,
+					"chunk_address", c.Address(),
+					"batch_id", hex.EncodeToString(c.Stamp().BatchID()),
+					"stamp_timestamp", binary.BigEndian.Uint64(c.Stamp().Timestamp()),
+					"bin", bin,
+				)
 				return 0, 0, errors.Join(chunkErr, err)
 			}
+			s.logger.Debug("pullsync reserve put ok",
+				"peer_address", peer,
+				"chunk_address", c.Address(),
+				"batch_id", hex.EncodeToString(c.Stamp().BatchID()),
+				"stamp_index", hex.EncodeToString(c.Stamp().Index()),
+				"stamp_timestamp", binary.BigEndian.Uint64(c.Stamp().Timestamp()),
+				"bin", bin,
+			)
 			chunksPut++
 		}
 	}
 
+	if chunkErr != nil {
+		s.logger.Debug("pullsync sync finished with chunk errors",
+			"error", chunkErr,
+			"peer_address", peer,
+			"bin", bin,
+			"start", start,
+			"offer_topmost", topmost,
+			"chunks_put", chunksPut,
+			"wanted", wanted,
+		)
+	}
+
 	return topmost, chunksPut, chunkErr
 }
 
@@ -424,9 +529,26 @@ func (s *Syncer) makeOffer(ctx context.Context, rn pb.Get) (*pb.Offer, []*storer
 	o := new(pb.Offer)
 	o.Topmost = top
 	o.Chunks = make([]*pb.Chunk, 0, len(bincs))
-	for _, v := range bincs {
+	for i, v := range bincs {
 		o.Chunks = append(o.Chunks, &pb.Chunk{Address: v.Address.Bytes(), Sum: v.Sum})
+		s.logger.Debug("pullsync offer chunk",
+			"bin", rn.Bin,
+			"start", rn.Start,
+			"offer_idx", i,
+			"bin_id", v.BinID,
+			"chunk_address", v.Address,
+			"batch_id", hex.EncodeToString(v.BatchID),
+			"stamp_hash", hex.EncodeToString(v.StampHash),
+			"sum", hex.EncodeToString(v.Sum),
+			"offer_topmost", top,
+		)
 	}
+	s.logger.Debug("pullsync offer summary",
+		"bin", rn.Bin,
+		"start", rn.Start,
+		"count", len(bincs),
+		"offer_topmost", top,
+	)
 	return o, bincs, nil
 }
 
@@ -465,7 +587,7 @@ func (s *Syncer) collectAddrs(ctx context.Context, bin uint8, start uint64) ([]*
 					break LOOP // The stream has been closed.
 				}
 
-				chs = append(chs, &storer.BinC{Address: c.Address, BatchID: c.BatchID, StampHash: c.StampHash, Sum: c.Sum})
+				chs = append(chs, &storer.BinC{Address: c.Address, BinID: c.BinID, BatchID: c.BatchID, StampHash: c.StampHash, Sum: c.Sum})
 				if c.BinID > topmost {
 					topmost = c.BinID
 				}
@@ -516,11 +638,34 @@ func (s *Syncer) processWant(ctx context.Context, bincs []*storer.BinC, w *pb.Wa
 			s.metrics.SentWanted.Inc()
 			ch, err := s.store.ReserveGet(ctx, c.Address, c.BatchID, c.StampHash)
 			if err != nil {
-				s.logger.Debug("processing want: unable to find chunk", "chunk_address", c.Address, "batch_id", hex.EncodeToString(c.BatchID))
+				s.logger.Debug("processing want: unable to find chunk",
+					"chunk_address", c.Address,
+					"batch_id", hex.EncodeToString(c.BatchID),
+					"stamp_hash", hex.EncodeToString(c.StampHash),
+					"bin_id", c.BinID,
+					"sum", hex.EncodeToString(c.Sum),
+					"offer_idx", i,
+				)
 				chunks = append(chunks, swarm.NewChunk(swarm.ZeroAddress, nil))
 				s.metrics.MissingChunks.Inc()
 				continue
 			}
+			stampTS := uint64(0)
+			stampIndex := ""
+			if ch.Stamp() != nil {
+				stampTS = binary.BigEndian.Uint64(ch.Stamp().Timestamp())
+				stampIndex = hex.EncodeToString(ch.Stamp().Index())
+			}
+			s.logger.Debug("pullsync deliver chunk",
+				"offer_idx", i,
+				"bin_id", c.BinID,
+				"chunk_address", c.Address,
+				"batch_id", hex.EncodeToString(c.BatchID),
+				"stamp_hash", hex.EncodeToString(c.StampHash),
+				"stamp_index", stampIndex,
+				"stamp_timestamp", stampTS,
+				"sum", hex.EncodeToString(c.Sum),
+			)
 			chunks = append(chunks, ch)
 		}
 	}

pkg/pushsync/pushsync.go

@@ -8,6 +8,8 @@ package pushsync
 
 import (
 	"context"
+	"encoding/binary"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"strconv"
@@ -266,13 +268,51 @@ func (ps *PushSync) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream)
 
 		chunkToPut, err := ps.validStamp(chunk)
 		if err != nil {
+			ps.logger.Debug("pushsync invalid stamp",
+				"error", err,
+				"peer_address", p.Address,
+				"chunk_address", chunkAddress,
+				"batch_id", hex.EncodeToString(chunk.Stamp().BatchID()),
+				"stamp_index", hex.EncodeToString(chunk.Stamp().Index()),
+				"stamp_timestamp", binary.BigEndian.Uint64(chunk.Stamp().Timestamp()),
+			)
 			return fmt.Errorf("invalid stamp: %w", err)
 		}
 
+		sumHex := ""
+		if sum, sumErr := storage.ChunkSum(chunkToPut); sumErr == nil {
+			sumHex = hex.EncodeToString(sum)
+		}
+		wrappedHex := ""
+		if sch, socErr := soc.FromChunk(chunkToPut); socErr == nil {
+			wrappedHex = sch.WrappedChunk().Address().String()
+		}
+
 		err = ps.store.ReservePutter().Put(ctx, chunkToPut)
 		if err != nil {
+			ps.logger.Debug("pushsync reserve put failed",
+				"error", err,
+				"peer_address", p.Address,
+				"chunk_address", chunkAddress,
+				"batch_id", hex.EncodeToString(chunkToPut.Stamp().BatchID()),
+				"stamp_index", hex.EncodeToString(chunkToPut.Stamp().Index()),
+				"stamp_timestamp", binary.BigEndian.Uint64(chunkToPut.Stamp().Timestamp()),
+				"sum", sumHex,
+				"wrapped_chunk_address", wrappedHex,
+				"path", reason,
+			)
 			return fmt.Errorf("reserve put: %w", err)
 		}
+		ps.logger.Debug("pushsync reserve put ok",
+			"peer_address", p.Address,
+			"chunk_address", chunkAddress,
+			"batch_id", hex.EncodeToString(chunkToPut.Stamp().BatchID()),
+			"stamp_index", hex.EncodeToString(chunkToPut.Stamp().Index()),
+			"stamp_timestamp", binary.BigEndian.Uint64(chunkToPut.Stamp().Timestamp()),
+			"sum", sumHex,
+			"wrapped_chunk_address", wrappedHex,
+			"path", reason,
+		)
 
 		signature, err := ps.signer.Sign(chunkToPut.Address().Bytes())
 		if err != nil {
@@ -307,6 +347,13 @@ func (ps *PushSync) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream)
 		return store(ctx)
 	case err == nil:
 		ps.metrics.Forwarder.Inc()
+		ps.logger.Debug("pushsync forward chunk",
+			"peer_address", p.Address,
+			"chunk_address", chunkAddress,
+			"batch_id", hex.EncodeToString(chunk.Stamp().BatchID()),
+			"stamp_index", hex.EncodeToString(chunk.Stamp().Index()),
+			"stamp_timestamp", binary.BigEndian.Uint64(chunk.Stamp().Timestamp()),
+		)
 
 		debit, err := ps.accounting.PrepareDebit(ctx, p.Address, price)
 		if err != nil {

pkg/storer/internal/reserve/convergence_test.go

@@ -330,6 +330,37 @@ func TestPutOrderConvergence(t *testing.T) {
 				}
 			},
 		},
+		{
+			// Same SOC address, same batch, equal timestamp, different stamp
+			// indices. putSOC treats this as a new stamp entry and replaces the
+			// shared payload unconditionally (last-write wins). Desired: settle
+			// on the lexicographically lower stamp hash like the same-slot path.
+			name:       "divergent socs, equal timestamp, distinct stamp indices",
+			unresolved: true,
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				return []swarm.Chunk{
+					newTestSOC(t, signer, id1, []byte("soc payload one")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 5)),
+					newTestSOC(t, signer, id1, []byte("soc payload two")).WithStamp(postagetesting.MustNewFields(batchA.ID, 1, 5)),
+				}
+			},
+		},
+		{
+			// Same SOC address under two batches at the same timestamp.
+			// putSOC currently replaces the shared payload on the second stamp
+			// unconditionally (last-write wins), so arrival order decides the
+			// payload. Desired: settle on the lexicographically lower stamp
+			// hash, matching the same-slot equal-timestamp path.
+			name:       "divergent socs, equal timestamp, distinct batches",
+			unresolved: true,
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				return []swarm.Chunk{
+					newTestSOC(t, signer, id1, []byte("soc payload one")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 5)),
+					newTestSOC(t, signer, id1, []byte("soc payload two")).WithStamp(postagetesting.MustNewFields(batchB.ID, 0, 5)),
+				}
+			},
+		},
 		{
 			// Three-way conflict across batches: the batch B entry must end
 			// serving whatever payload the batch A conflict settles on, with