feat: pullsync chuns convergence

config/config.yaml

@@ -355,6 +355,17 @@ checks:
       request-timeout: 5m
     timeout: 5m
     type: soc
+  socmatrix:
+    options:
+      postage-ttl: 24h
+      postage-depth: 21
+      postage-label: socmatrix-label
+      sync-wait: 45s
+      sync-retry-interval: 2s
+      request-timeout: 15m
+      password: beekeeper
+    timeout: 20m
+    type: socmatrix
   stake:
     type: stake
     timeout: 5m

config/local.yaml

@@ -382,6 +382,17 @@ checks:
       request-timeout: 5m
     timeout: 5m
     type: soc
+  ci-socmatrix:
+    options:
+      postage-ttl: 24h
+      postage-depth: 21
+      postage-label: socmatrix-label
+      sync-wait: 45s # wait after upload for push/pull sync before neighborhood assert (precision 11–16)
+      sync-retry-interval: 2s
+      request-timeout: 45m
+      password: beekeeper
+    timeout: 60m
+    type: socmatrix
   ci-content-availability:
     type: content-availability
     timeout: 5m

pkg/bee/api/api.go

@@ -17,6 +17,7 @@ const (
 	apiVersion                  = "v1"
 	contentType                 = "application/json, text/plain, */*; charset=utf-8"
 	postageStampBatchHeader     = "Swarm-Postage-Batch-Id"
+	postageStampHeader          = "Swarm-Postage-Stamp"
 	deferredUploadHeader        = "Swarm-Deferred-Upload"
 	swarmAct                    = "Swarm-Act"
 	swarmActHistoryAddress      = "Swarm-Act-History-Address"
@@ -46,6 +47,7 @@ type Client struct {
 	Bytes       *BytesService
 	Chunks      *ChunksService
 	Dirs        *DirsService
+	Envelope    *EnvelopeService
 	Feed        *FeedService
 	Files       *FilesService
 	Node        *NodeService
@@ -96,6 +98,7 @@ func newClient(apiURL *url.URL, httpClient *http.Client) (c *Client) {
 	c.Bytes = (*BytesService)(&c.service)
 	c.Chunks = (*ChunksService)(&c.service)
 	c.Dirs = (*DirsService)(&c.service)
+	c.Envelope = (*EnvelopeService)(&c.service)
 	c.Feed = (*FeedService)(&c.service)
 	c.Files = (*FilesService)(&c.service)
 	c.Node = (*NodeService)(&c.service)

pkg/bee/api/chunks.go

@@ -36,7 +36,11 @@ func (c *ChunksService) Upload(ctx context.Context, data []byte, o UploadOptions
 	if o.Direct {
 		h.Add(deferredUploadHeader, "false")
 	}
-	h.Add(postageStampBatchHeader, o.BatchID)
+	if o.Stamp != "" {
+		h.Add(postageStampHeader, o.Stamp)
+	} else {
+		h.Add(postageStampBatchHeader, o.BatchID)
+	}
 	err := c.client.requestWithHeader(ctx, http.MethodPost, "/"+apiVersion+"/chunks", h, bytes.NewReader(data), &resp)
 	return resp, err
 }

pkg/bee/api/envelope.go

@@ -0,0 +1,53 @@
+package api
+
+import (
+	"context"
+	"encoding/hex"
+	"fmt"
+	"net/http"
+
+	"github.com/ethersphere/bee/v2/pkg/postage"
+	"github.com/ethersphere/bee/v2/pkg/swarm"
+)
+
+// EnvelopeService represents Bee's postage envelope service.
+type EnvelopeService service
+
+// EnvelopeResponse is the JSON body returned by POST /envelope/{address}.
+type EnvelopeResponse struct {
+	Issuer    string `json:"issuer"`
+	Index     string `json:"index"`
+	Timestamp string `json:"timestamp"`
+	Signature string `json:"signature"`
+}
+
+// Create requests a postage stamp for the given chunk address from the node's
+// stamper for batchID. The stamp is returned as a postage.Stamp ready for upload.
+func (e *EnvelopeService) Create(ctx context.Context, addr swarm.Address, batchID string) (*postage.Stamp, error) {
+	h := http.Header{}
+	h.Add(postageStampBatchHeader, batchID)
+
+	var resp EnvelopeResponse
+	err := e.client.requestWithHeader(ctx, http.MethodPost, "/"+apiVersion+"/envelope/"+addr.String(), h, nil, &resp)
+	if err != nil {
+		return nil, err
+	}
+
+	batchBytes, err := hex.DecodeString(batchID)
+	if err != nil {
+		return nil, fmt.Errorf("decode batch id: %w", err)
+	}
+	index, err := hex.DecodeString(resp.Index)
+	if err != nil {
+		return nil, fmt.Errorf("decode stamp index: %w", err)
+	}
+	ts, err := hex.DecodeString(resp.Timestamp)
+	if err != nil {
+		return nil, fmt.Errorf("decode stamp timestamp: %w", err)
+	}
+	sig, err := hex.DecodeString(resp.Signature)
+	if err != nil {
+		return nil, fmt.Errorf("decode stamp signature: %w", err)
+	}
+	return postage.NewStamp(batchBytes, index, ts, sig), nil
+}

pkg/bee/api/node.go

@@ -80,19 +80,21 @@ func (n *NodeService) Balances(ctx context.Context) (resp Balances, err error) {
 
 // HasChunk returns true/false if node has a chunk
 func (n *NodeService) HasChunk(ctx context.Context, a swarm.Address) (bool, error) {
-	resp := struct {
-		Message string `json:"message,omitempty"`
-		Code    int    `json:"code,omitempty"`
-	}{}
-
-	err := n.client.requestJSON(ctx, http.MethodGet, "/chunks/"+a.String(), nil, &resp)
-	if IsHTTPStatusErrorCode(err, http.StatusNotFound) {
-		return false, nil
-	} else if err != nil {
-		return false, err
+	for attempt := 0; attempt < 5; attempt++ {
+		err := n.client.requestJSON(ctx, http.MethodHead, "/chunks/"+a.String(), nil, nil)
+		if IsHTTPStatusErrorCode(err, http.StatusNotFound) {
+			return false, nil
+		}
+		if IsHTTPStatusErrorCode(err, http.StatusServiceUnavailable) {
+			time.Sleep(1 * time.Second)
+			continue
+		}
+		if err != nil {
+			return false, err
+		}
+		return true, nil
 	}
-
-	return true, nil
+	return false, fmt.Errorf("has chunk %s: exceeded retries on 503", a.String())
 }
 
 // Health represents node's health

pkg/bee/api/options.go

@@ -10,6 +10,7 @@ type UploadOptions struct {
 	Pin               bool
 	Tag               uint64
 	BatchID           string
+	Stamp             string // hex-encoded postage stamp; when set, BatchID is ignored
 	Direct            bool
 	ActHistoryAddress swarm.Address
 	RLevel            *redundancy.Level

pkg/bee/api/soc.go

@@ -9,17 +9,32 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 )
 
-// PSSService represents Bee's PSS service
+// SOCService represents Bee's SOC service.
 type SOCService service
 
+// SocResponse is the JSON body returned by POST /soc.
 type SocResponse struct {
-	Reference swarm.Address
+	Reference swarm.Address `json:"reference"`
 }
 
-// Sends a PSS message to a recipienct with a specific topic
+// UploadSOC uploads a single owner chunk using a postage batch ID.
 func (p *SOCService) UploadSOC(ctx context.Context, owner, ID, signature string, data io.Reader, batchID string) (*SocResponse, error) {
 	h := http.Header{}
 	h.Add(postageStampBatchHeader, batchID)
+	h.Add(deferredUploadHeader, "false")
+	url := fmt.Sprintf("/%s/soc/%s/%s?sig=%s", apiVersion, owner, ID, signature)
+
+	resp := SocResponse{}
+	return &resp, p.client.requestWithHeader(ctx, http.MethodPost, url, h, data, &resp)
+}
+
+// UploadSOCWithStamp uploads a single owner chunk using a precomputed postage stamp
+// (hex-encoded Swarm-Postage-Stamp header). Direct upload is forced so the chunk
+// enters the reserve immediately for pullsync.
+func (p *SOCService) UploadSOCWithStamp(ctx context.Context, owner, ID, signature string, data io.Reader, stampHex string) (*SocResponse, error) {
+	h := http.Header{}
+	h.Add(postageStampHeader, stampHex)
+	h.Add(deferredUploadHeader, "false")
 	url := fmt.Sprintf("/%s/soc/%s/%s?sig=%s", apiVersion, owner, ID, signature)
 
 	resp := SocResponse{}

pkg/bee/client.go

@@ -3,6 +3,7 @@ package bee
 import (
 	"bytes"
 	"context"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -12,11 +13,13 @@ import (
 	"net/http"
 	"net/url"
 	"slices"
+	"strings"
 	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethersphere/bee/v2/pkg/crypto"
+	"github.com/ethersphere/bee/v2/pkg/postage"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	"github.com/ethersphere/beekeeper/pkg/bee/api"
 	"github.com/ethersphere/beekeeper/pkg/logging"
@@ -213,13 +216,19 @@ func (c *Client) DownloadBytes(ctx context.Context, a swarm.Address, opts *api.D
 
 // DownloadChunk downloads chunk from the node
 func (c *Client) DownloadChunk(ctx context.Context, a swarm.Address, targets string, opts *api.DownloadOptions) (data []byte, err error) {
-	r, err := c.api.Chunks.Download(ctx, a, targets, opts)
-	if err != nil {
-		return nil, fmt.Errorf("download chunk %s: %w", a, err)
+	for attempt := 0; attempt < 5; attempt++ {
+		r, err := c.api.Chunks.Download(ctx, a, targets, opts)
+		if err != nil {
+			if api.IsHTTPStatusErrorCode(err, 503) || strings.Contains(err.Error(), "503") || strings.Contains(err.Error(), "Service Unavailable") {
+				time.Sleep(1 * time.Second)
+				continue
+			}
+			return nil, fmt.Errorf("download chunk %s: %w", a, err)
+		}
+		defer r.Close()
+		return io.ReadAll(r)
 	}
-	defer r.Close()
-
-	return io.ReadAll(r)
+	return nil, fmt.Errorf("download chunk %s: exceeded retries on 503", a)
 }
 
 // DownloadFileBytes downloads a flie from the node and returns the data.
@@ -642,7 +651,7 @@ func (c *Client) SendPSSMessage(ctx context.Context, nodeAddress swarm.Address,
 	return c.api.PSS.SendMessage(ctx, nodeAddress, publicKey, topic, prefix, bytes.NewReader(data), batchID)
 }
 
-// UploadSOC uploads a single owner chunk to a node with a E
+// UploadSOC uploads a single owner chunk to a node with a batch ID.
 func (c *Client) UploadSOC(ctx context.Context, owner, ID, signature string, data []byte, batchID string) (swarm.Address, error) {
 	resp, err := c.api.SOC.UploadSOC(ctx, owner, ID, signature, bytes.NewReader(data), batchID)
 	if err != nil {
@@ -652,6 +661,21 @@ func (c *Client) UploadSOC(ctx context.Context, owner, ID, signature string, dat
 	return resp.Reference, nil
 }
 
+// UploadSOCWithStamp uploads a single owner chunk using a precomputed postage stamp.
+func (c *Client) UploadSOCWithStamp(ctx context.Context, owner, ID, signature string, data []byte, stamp []byte) (swarm.Address, error) {
+	resp, err := c.api.SOC.UploadSOCWithStamp(ctx, owner, ID, signature, bytes.NewReader(data), hex.EncodeToString(stamp))
+	if err != nil {
+		return swarm.ZeroAddress, err
+	}
+
+	return resp.Reference, nil
+}
+
+// CreateEnvelope requests a postage stamp for addr from the node's stamper.
+func (c *Client) CreateEnvelope(ctx context.Context, addr swarm.Address, batchID string) (*postage.Stamp, error) {
+	return c.api.Envelope.Create(ctx, addr, batchID)
+}
+
 // Settlements represents Settlements's response
 type Settlements struct {
 	Settlements   []Settlement