diff --git a/cli/config/config.go b/cli/config/config.go index 6489da134d..0e0e946513 100644 --- a/cli/config/config.go +++ b/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", diff --git a/cli/p2p_document_sync.go b/cli/p2p_document_sync.go index 303a9ce791..7088116b72 100644 --- a/cli/p2p_document_sync.go +++ b/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 } diff --git a/cli/start.go b/cli/start.go index 8b3f2a484b..0913283a61 100644 --- a/cli/start.go +++ b/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"]), diff --git a/client/options/p2p.go b/client/options/p2p.go index 5016aa0418..273959ede5 100644 --- a/client/options/p2p.go +++ b/client/options/p2p.go @@ -11,6 +11,8 @@ package options import ( + "time" + "github.com/sourcenetwork/immutable" "github.com/sourcenetwork/defradb/acp/identity" @@ -482,6 +484,13 @@ 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. @@ -489,6 +498,11 @@ 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 +} diff --git a/docs/website/references/cli/defradb_client_p2p_document_sync.md b/docs/website/references/cli/defradb_client_p2p_document_sync.md index 1b359a3553..af30dffbf7 100644 --- a/docs/website/references/cli/defradb_client_p2p_document_sync.md +++ 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 diff --git a/docs/website/references/cli/defradb_start.md b/docs/website/references/cli/defradb_start.md index 8dbbd182bb..3ec127121b 100644 --- a/docs/website/references/cli/defradb_start.md +++ b/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 diff --git a/docs/website/references/http/openapi.json b/docs/website/references/http/openapi.json index 5622922f99..56d793350c 100644 --- a/docs/website/references/http/openapi.json +++ b/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" } } }, diff --git a/http/client_p2p.go b/http/client_p2p.go index 8a31b8df60..6e8339ccd7 100644 --- a/http/client_p2p.go +++ b/http/client_p2p.go @@ -41,6 +41,19 @@ type DeleteReplicatorParams struct { 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 @@ func (c *Client) SyncDocuments( 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 } diff --git a/http/handler_p2p.go b/http/handler_p2p.go index 137605a094..c38abebfbd 100644 --- a/http/handler_p2p.go +++ b/http/handler_p2p.go @@ -256,11 +256,7 @@ func (h *p2pHandler) ListP2PDocuments(rw http.ResponseWriter, req *http.Request) 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 @@ func (h *p2pHandler) SyncDocuments(rw http.ResponseWriter, req *http.Request) { } 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 @@ func (h *p2pHandler) bindRoutes(router *Router) { 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") diff --git a/http/openapi.go b/http/openapi.go index 403f711a1a..9f3d53e2be 100644 --- a/http/openapi.go +++ b/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{}, diff --git a/internal/db/config.go b/internal/db/config.go index 7a6b1d7444..6afd27985d 100644 --- a/internal/db/config.go +++ b/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, }, } } diff --git a/internal/db/p2p.go b/internal/db/p2p.go index a531142c8c..1de3c42ea3 100644 --- a/internal/db/p2p.go +++ b/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 } diff --git a/internal/db/p2p/errors.go b/internal/db/p2p/errors.go index 576074fb54..e5f9a34856 100644 --- a/internal/db/p2p/errors.go +++ b/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) } diff --git a/internal/db/p2p/sync_dag.go b/internal/db/p2p/sync_dag.go index d92d5b3e77..b5104658f0 100644 --- a/internal/db/p2p/sync_dag.go +++ b/internal/db/p2p/sync_dag.go @@ -12,6 +12,7 @@ package p2p import ( "context" + "time" "github.com/ipld/go-ipld-prime/linking" cidlink "github.com/ipld/go-ipld-prime/linking/cid" @@ -19,11 +20,31 @@ import ( "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) } diff --git a/internal/db/p2p/sync_dag_test.go b/internal/db/p2p/sync_dag_test.go new file mode 100644 index 0000000000..f62b944476 --- /dev/null +++ b/internal/db/p2p/sync_dag_test.go @@ -0,0 +1,37 @@ +// Copyright 2026 Democratized Data Foundation +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +package p2p + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +// The per-block fetch timeout should default to the node setting, and a positive per-request +// override carried on the context should take precedence. A zero or negative override is ignored +// so a caller cannot accidentally disable the timeout. +func TestBlockSyncTimeout_OverrideResolution(t *testing.T) { + p := &P2P{syncBlockLinkTimeout: 5 * time.Second} + + assert.Equal(t, 5*time.Second, p.blockSyncTimeout(context.Background()), + "with no override the node default should be used") + + overridden := WithBlockSyncTimeout(context.Background(), 30*time.Second) + assert.Equal(t, 30*time.Second, p.blockSyncTimeout(overridden), + "a positive override should take precedence over the node default") + + zeroOverride := WithBlockSyncTimeout(context.Background(), 0) + assert.Equal(t, 5*time.Second, p.blockSyncTimeout(zeroOverride), + "a non-positive override should be ignored in favour of the node default") +} diff --git a/node/node.go b/node/node.go index fa8b38f83e..506a108aa6 100644 --- a/node/node.go +++ b/node/node.go @@ -117,7 +117,7 @@ func DefaultNodeOptions() options.NodeOptions { time.Minute * 16, time.Minute * 32, }, - P2PBlockSyncTimeout: time.Second * 5, + P2PBlockSyncTimeout: time.Second * 30, LensRuntime: options.NodeDefaultLensRuntime, }, P2P: options.NodeP2POptions{}, diff --git a/tests/clients/cli/wrapper.go b/tests/clients/cli/wrapper.go index 1cbfde69b0..cf4f568521 100644 --- a/tests/clients/cli/wrapper.go +++ b/tests/clients/cli/wrapper.go @@ -314,6 +314,9 @@ func (w *Wrapper) SyncDocuments( if hasDeadline { args = append(args, "--timeout", time.Until(deadline).String()) } + if blockSyncTimeout := opt.GetBlockSyncTimeout(); blockSyncTimeout.HasValue() { + args = append(args, "--block-sync-timeout", blockSyncTimeout.Value().String()) + } args = append(args, collectionName) args = append(args, docIDs...) diff --git a/tests/integration/db.go b/tests/integration/db.go index 49e9dd58d1..07f512e405 100644 --- a/tests/integration/db.go +++ b/tests/integration/db.go @@ -91,8 +91,9 @@ func defaultNodeOpts() *options.NodeOptionsBuilder { opt.DB(). SetLensPoolSize(lensPoolSize). SetLensRuntime(lensType). - // The default is 5 and that is never going to be needed in a testing scenario where all the - // nodes are on the same machine with no network latency. + // The production default (30s) is never going to be needed in a testing scenario where all + // the nodes are on the same machine with no network latency; a short timeout keeps tests + // that exercise the timeout path fast. SetP2PBlockSyncTimeout(1 * time.Second) return opt