feat: Configurable P2P block-sync timeout

cli/config/config.go

@@ -74,6 +74,7 @@ var ConfigFlags = map[string]string{
 	"no-p2p":                     "net.p2pdisabled",
 	"pubsub":                     "net.pubsubenabled",
 	"relay":                      "net.relay",
+	"p2p-block-sync-timeout":     "net.p2pblocksynctimeout",
 	"allowed-origins":            "api.allowed-origins",
 	"pubkeypath":                 "api.pubkeypath",
 	"privkeypath":                "api.privkeypath",
@@ -104,6 +105,7 @@ var ConfigDefaults = map[string]any{
 	"net.peers":                         []string{},
 	"net.pubSubEnabled":                 true,
 	"net.relay":                         false,
+	"net.p2pblocksynctimeout":           30,
 	"keyring.backend":                   "file",
 	"keyring.disabled":                  false,
 	"keyring.namespace":                 "defradb",

cli/p2p_document_sync.go

@@ -41,6 +41,9 @@ It doesn't automatically subscribe to the collection or the documents.`,
 
 			cliClient := mustGetContextCLIClient(cmd)
 			opt := options.WithIdentity(options.SyncDocuments(), iIdentity.FromContext(cmd.Context()))
+			if blockSyncTimeout, _ := cmd.Flags().GetDuration("block-sync-timeout"); blockSyncTimeout > 0 {
+				opt = opt.SetBlockSyncTimeout(blockSyncTimeout)
+			}
 			return cliClient.SyncDocuments(ctx, collectionName, docIDs, opt)
 		},
 	}
@@ -51,6 +54,8 @@ It doesn't automatically subscribe to the collection or the documents.`,
 	EmbedCLIExample(ctx, cmd, "sync multiple documents",
 		`defradb client p2p document sync Users bae123 bae456`)
 
-	cmd.Flags().Duration("timeout", 0, "Timeout for sync operations")
+	cmd.Flags().Duration("timeout", 0, "Timeout for the whole sync operation")
+	cmd.Flags().Duration("block-sync-timeout", 0,
+		"Per-block fetch timeout for this sync, overriding the node default (e.g. 30s)")
 	return cmd
 }

cli/start.go

@@ -99,6 +99,9 @@ func MakeStartCommand(ctx context.Context) *cobra.Command {
 				SetMaxTxnRetries(cfg.GetInt("datastore.MaxTxnRetries")).
 				SetRetryIntervals(replicatorRetryIntervals).
 				SetLensRuntime(options.NodeLensRuntimeType(cfg.GetString("lens.runtime")))
+			if p2pBlockSyncTimeout := cfg.GetInt("net.p2pblocksynctimeout"); p2pBlockSyncTimeout > 0 {
+				opts.DB().SetP2PBlockSyncTimeout(time.Duration(p2pBlockSyncTimeout) * time.Second)
+			}
 			opts.P2P().
 				SetListenAddresses(cfg.GetStringSlice("net.p2pAddresses")...).
 				SetEnablePubSub(cfg.GetBool("net.pubSubEnabled")).
@@ -314,6 +317,11 @@ func MakeStartCommand(ctx context.Context) *cobra.Command {
 		cfg.GetBool(config.ConfigFlags["relay"]),
 		"Enable the p2p relay",
 	)
+	cmd.PersistentFlags().Int(
+		"p2p-block-sync-timeout",
+		cfg.GetInt(config.ConfigFlags["p2p-block-sync-timeout"]),
+		"Timeout in seconds for fetching each block during P2P DAG sync",
+	)
 	cmd.PersistentFlags().StringArray(
 		"allowed-origins",
 		cfg.GetStringSlice(config.ConfigFlags["allowed-origins"]),

client/options/p2p.go

@@ -11,6 +11,8 @@
 package options
 
 import (
+	"time"
+
 	"github.com/sourcenetwork/immutable"
 
 	"github.com/sourcenetwork/defradb/acp/identity"
@@ -482,13 +484,25 @@ func (b *ListP2PDocumentsOptionsBuilder) SetIdentity(id identity.Identity) *List
 type SyncDocumentsOptions struct {
 	// Identity is the identity of the actor performing the operation.
 	Identity immutable.Option[identity.Identity]
+
+	// BlockSyncTimeout, when set, overrides the node's default per-block fetch timeout for this
+	// sync only. It bounds how long the node waits for each linked block to arrive from a peer;
+	// a peer that is slow to authorize or serve a block past this budget causes the sync to fail
+	// with a block-sync timeout. It does not bound the overall operation — use a context deadline
+	// for that.
+	BlockSyncTimeout immutable.Option[time.Duration]
 }
 
 // GetIdentity returns the identity for the operation.
 func (o *SyncDocumentsOptions) GetIdentity() immutable.Option[identity.Identity] {
 	return o.Identity
 }
 
+// GetBlockSyncTimeout returns the per-block fetch timeout override for the operation, if set.
+func (o *SyncDocumentsOptions) GetBlockSyncTimeout() immutable.Option[time.Duration] {
+	return o.BlockSyncTimeout
+}
+
 // SyncDocumentsOptionsBuilder is a builder for SyncDocumentsOptions.
 type SyncDocumentsOptionsBuilder struct {
 	enumerableBuilder[SyncDocumentsOptions]
@@ -506,3 +520,11 @@ func (b *SyncDocumentsOptionsBuilder) SetIdentity(id identity.Identity) *SyncDoc
 	})
 	return b
 }
+
+// SetBlockSyncTimeout overrides the node's default per-block fetch timeout for this sync.
+func (b *SyncDocumentsOptionsBuilder) SetBlockSyncTimeout(timeout time.Duration) *SyncDocumentsOptionsBuilder {
+	b.append(func(opts *SyncDocumentsOptions) {
+		opts.BlockSyncTimeout = immutable.Some(timeout)
+	})
+	return b
+}

docs/website/references/cli/defradb_client_p2p_document_sync.md

@@ -26,8 +26,9 @@ sync multiple documents:
 ### Options
 
 ```
-  -h, --help               help for sync
-      --timeout duration   Timeout for sync operations
+      --block-sync-timeout duration   Per-block fetch timeout for this sync, overriding the node default (e.g. 30s)
+  -h, --help                          help for sync
+      --timeout duration              Timeout for the whole sync operation
 ```
 
 ### Options inherited from parent commands

docs/website/references/cli/defradb_start.md

@@ -33,6 +33,7 @@ defradb start [flags]
       --no-signing                        Disable signing of commits.
       --no-telemetry                      Disables telemetry reporting. Telemetry is only enabled in builds that use the telemetry flag.
       --node-acp-enable                   Enable the node access control system.
+      --p2p-block-sync-timeout int        Timeout in seconds for fetching each block during P2P DAG sync (default 30)
       --p2paddr strings                   Listen addresses for the p2p network (formatted as a libp2p MultiAddr) (default [/ip4/127.0.0.1/tcp/9171])
       --peers stringArray                 List of peers to connect to
       --privkeypath string                Path to the private key for tls

docs/website/references/http/openapi.json

@@ -627,6 +627,26 @@
                 },
                 "type": "object"
             },
+            "sync_documents_params": {
+                "properties": {
+                    "blockSyncTimeout": {
+                        "type": "string"
+                    },
+                    "collectionName": {
+                        "type": "string"
+                    },
+                    "docIDs": {
+                        "items": {
+                            "type": "string"
+                        },
+                        "type": "array"
+                    },
+                    "timeout": {
+                        "type": "string"
+                    }
+                },
+                "type": "object"
+            },
             "update_collection": {
                 "properties": {
                     "filter": {},
@@ -2511,21 +2531,7 @@
                     "content": {
                         "application/json": {
                             "schema": {
-                                "properties": {
-                                    "collectionName": {
-                                        "type": "string"
-                                    },
-                                    "docIDs": {
-                                        "items": {
-                                            "type": "string"
-                                        },
-                                        "type": "array"
-                                    },
-                                    "timeout": {
-                                        "type": "string"
-                                    }
-                                },
-                                "type": "object"
+                                "$ref": "#/components/schemas/sync_documents_params"
                             }
                         }
                     },

http/client_p2p.go

@@ -41,6 +41,19 @@
 	Collections []string
 }
 
+// SyncDocumentsParams contains the params for the sync documents request.
+type SyncDocumentsParams struct {
+	// CollectionName is the name of the collection containing the documents to sync.
+	CollectionName string `json:"collectionName"`
+	// DocIDs are the IDs of the documents to sync.
+	DocIDs []string `json:"docIDs"`
+	// Timeout, when set, bounds the whole sync operation (as a duration string, e.g. "10s").
+	Timeout string `json:"timeout,omitempty"`
+	// BlockSyncTimeout, when set, overrides the node's default per-block fetch timeout for this
+	// sync only (as a duration string, e.g. "30s").
+	BlockSyncTimeout string `json:"blockSyncTimeout,omitempty"`
+}
+
 func (c *Client) PeerInfo(ctx context.Context, opts ...options.Enumerable[options.PeerInfoOptions]) ([]string, error) {
 	opt := utils.NewOptions(opts...)
 	ctx = identity.WithContext(ctx, opt.GetIdentity())
@@ -331,16 +344,20 @@
 
 	methodURL := c.http.apiURL.JoinPath("p2p", "documents", "sync")
 
-	req := map[string]any{
-		"collectionName": collectionName,
-		"docIDs":         docIDs,
+	params := SyncDocumentsParams{
+		CollectionName: collectionName,
+		DocIDs:         docIDs,
+	}
+
+	if blockSyncTimeout := opt.GetBlockSyncTimeout(); blockSyncTimeout.HasValue() {
+		params.BlockSyncTimeout = blockSyncTimeout.Value().String()
 	}
 
 	deadline, hasDeadline := ctx.Deadline()
 	if hasDeadline {
-		req["timeout"] = time.Until(deadline).String()
+		params.Timeout = time.Until(deadline).String()
 	}
-	body, err := json.Marshal(req)
+	body, err := json.Marshal(params)
 	if err != nil {
 		return err
 	}

http/handler_p2p.go

@@ -256,11 +256,7 @@
 func (h *p2pHandler) SyncDocuments(rw http.ResponseWriter, req *http.Request) {
 	db := mustGetContextClientDB(req)
 
-	var reqBody struct {
-		CollectionName string   `json:"collectionName"`
-		DocIDs         []string `json:"docIDs"`
-		Timeout        string   `json:"timeout"`
-	}
+	var reqBody SyncDocumentsParams
 
 	if err := requestJSON(req, &reqBody); err != nil {
 		responseJSON(rw, http.StatusBadRequest, errorResponse{err})
@@ -280,6 +276,14 @@
 	}
 
 	opts := options.WithIdentity(options.SyncDocuments(), identity.FromContext(ctx))
+	if reqBody.BlockSyncTimeout != "" {
+		blockSyncTimeout, err := time.ParseDuration(reqBody.BlockSyncTimeout)
+		if err != nil {
+			responseJSON(rw, http.StatusBadRequest, errorResponse{err})
+			return
+		}
+		opts = opts.SetBlockSyncTimeout(blockSyncTimeout)
+	}
 	err := db.SyncDocuments(ctx, reqBody.CollectionName, reqBody.DocIDs, opts)
 	if err != nil {
 		responseJSON(rw, http.StatusInternalServerError, errorResponse{err})
@@ -539,14 +543,12 @@
 	deletePeerDocuments.Responses.Set("200", successResponse)
 	deletePeerDocuments.Responses.Set("400", errorResponse)
 
-	syncDocumentsRequestSchema := openapi3.NewObjectSchema().
-		WithProperty("collectionName", openapi3.NewStringSchema()).
-		WithProperty("docIDs", openapi3.NewArraySchema().WithItems(openapi3.NewStringSchema())).
-		WithProperty("timeout", openapi3.NewStringSchema())
-
+	syncDocumentsParamsSchema := &openapi3.SchemaRef{
+		Ref: "#/components/schemas/sync_documents_params",
+	}
 	syncDocumentsRequest := openapi3.NewRequestBody().
 		WithRequired(true).
-		WithContent(openapi3.NewContentWithJSONSchema(syncDocumentsRequestSchema))
+		WithContent(openapi3.NewContentWithJSONSchemaRef(syncDocumentsParamsSchema))
 
 	syncDocumentsResponse := openapi3.NewResponse().
 		WithDescription("Document sync completed successfully")

http/openapi.go

@@ -41,6 +41,7 @@ var openApiSchemas = map[string]any{
 	"replicator":                               &client.Replicator{},
 	"add_replicator_params":                    &AddReplicatorParams{},
 	"delete_replicator_params":                 &DeleteReplicatorParams{},
+	"sync_documents_params":                    &SyncDocumentsParams{},
 	"ccip_request":                             &CCIPRequest{},
 	"ccip_response":                            &CCIPResponse{},
 	"patch_collection_request":                 &patchCollectionRequest{},

internal/db/config.go

@@ -39,7 +39,7 @@ func defaultDBConfig() intOpts.DBOptions {
 				time.Minute * 16,
 				time.Minute * 32,
 			},
-			P2PBlockSyncTimeout: time.Second * 5,
+			P2PBlockSyncTimeout: time.Second * 30,
 		},
 	}
 }

internal/db/p2p.go

@@ -17,6 +17,7 @@ import (
 	"github.com/sourcenetwork/defradb/client"
 	"github.com/sourcenetwork/defradb/client/options"
 	"github.com/sourcenetwork/defradb/event"
+	"github.com/sourcenetwork/defradb/internal/db/p2p"
 	"github.com/sourcenetwork/defradb/internal/identity"
 	"github.com/sourcenetwork/defradb/internal/utils"
 )
@@ -369,6 +370,10 @@ func (db *DB) SyncDocuments(
 
 	ctx = identity.WithContext(ctx, opt.Identity)
 
+	if opt.BlockSyncTimeout.HasValue() {
+		ctx = p2p.WithBlockSyncTimeout(ctx, opt.BlockSyncTimeout.Value())
+	}
+
 	if db.p2p == nil {
 		return ErrNoP2P
 	}

internal/db/p2p/errors.go

@@ -72,6 +72,7 @@ var (
 	ErrCollectionNotBranchable     = errors.New("collection is not branchable")
 	ErrNoHeadsForBranchableCol     = errors.New("no heads found for branchable collection")
 	ErrBlockCIDMismatch            = errors.New("pushed block does not match the advertised CID")
+	ErrBlockSyncTimeout            = errors.New("timeout while fetching linked block during DAG sync")
 )
 
 func NewErrReplicatorCollections(inner error, kv ...errors.KV) error {
@@ -162,6 +163,13 @@ func NewErrCheckBlockMerged(inner error) error   { return errors.Wrap(errCheckBl
 func NewErrVerifyBlockSig(inner error) error     { return errors.Wrap(errVerifyBlockSig, inner) }
 func NewErrGetEncKeysForBlock(inner error) error { return errors.Wrap(errGetEncKeysForBlock, inner) }
 func NewErrLoadLinkedBlock(inner error) error    { return errors.Wrap(errLoadLinkedBlock, inner) }
+
+// NewErrBlockSyncTimeout wraps the timeout error with the link that could not be fetched in time.
+// The result matches both [ErrBlockSyncTimeout] and the underlying cause under errors.Is.
+func NewErrBlockSyncTimeout(inner error, link string) error {
+	return errors.Wrap(ErrBlockSyncTimeout.Error(), errors.Join(ErrBlockSyncTimeout, inner), errors.NewKV("Link", link))
+}
+
 func NewErrDecodeLinkedBlock(inner error) error  { return errors.Wrap(errDecodeLinkedBlock, inner) }
 func NewErrProcessLinkedBlock(inner error) error { return errors.Wrap(errProcessLinkedBlock, inner) }
 func NewErrRetrieveEncKey(inner error) error     { return errors.Wrap(errRetrieveEncKey, inner) }

internal/db/p2p/sync_dag.go

@@ -12,18 +12,39 @@ package p2p
 
 import (
 	"context"
+	"time"
 
 	"github.com/ipld/go-ipld-prime/linking"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
 
 	"github.com/sourcenetwork/corekv/blockstore"
 	"github.com/sourcenetwork/immutable"
 
+	"github.com/sourcenetwork/defradb/errors"
 	coreblock "github.com/sourcenetwork/defradb/internal/core/block"
 	"github.com/sourcenetwork/defradb/internal/datastore"
 	"github.com/sourcenetwork/defradb/internal/encryption"
 )
 
+// blockSyncTimeoutCtxKey is the context key under which a per-request per-block sync timeout
+// override is carried down to loadBlockLinks.
+type blockSyncTimeoutCtxKey struct{}
+
+// WithBlockSyncTimeout returns a context carrying a per-block sync timeout override that takes
+// precedence over the node default for the DAG sync it drives.
+func WithBlockSyncTimeout(ctx context.Context, timeout time.Duration) context.Context {
+	return context.WithValue(ctx, blockSyncTimeoutCtxKey{}, timeout)
+}
+
+// blockSyncTimeout returns the per-block fetch timeout to use: the per-request override carried
+// on ctx if one was set (and positive), otherwise the node default.
+func (p *P2P) blockSyncTimeout(ctx context.Context) time.Duration {
+	if v, ok := ctx.Value(blockSyncTimeoutCtxKey{}).(time.Duration); ok && v > 0 {
+		return v
+	}
+	return p.syncBlockLinkTimeout
+}
+
 func makeLinkSystem(blockService blockstore.IPLDStore) linking.LinkSystem {
 	linkSys := cidlink.DefaultLinkSystem()
 	linkSys.SetWriteStorage(blockService)
@@ -99,11 +120,17 @@ func (p *P2P) loadBlockLinks(ctx context.Context, linkSys *linking.LinkSystem, b
 			return ctx.Err()
 		}
 
-		ctxWithTimeout, cancel := context.WithTimeout(ctx, p.syncBlockLinkTimeout)
+		ctxWithTimeout, cancel := context.WithTimeout(ctx, p.blockSyncTimeout(ctx))
 		nd, err := linkSys.Load(linking.LinkContext{Ctx: ctxWithTimeout}, lnk, coreblock.BlockSchemaPrototype)
 		cancel()
 
 		if err != nil {
+			// Distinguish "the peer did not serve this block in time" from other load failures.
+			// Only the per-block timeout is attributed here; a deadline on the parent ctx is a
+			// caller-level cancellation and is reported as-is.
+			if errors.Is(err, context.DeadlineExceeded) && ctx.Err() == nil {
+				return NewErrBlockSyncTimeout(err, lnk.String())
+			}
 			return NewErrLoadLinkedBlock(err)
 		}