From 7f20c1598ab56e3a6c3348d6d35e99ebe5f24bea Mon Sep 17 00:00:00 2001 From: Islam Aleiv Date: Thu, 9 Jul 2026 12:27:17 +0200 Subject: [PATCH 1/7] feat: Configurable and per-request P2P block-sync timeout Make the per-block DAG-sync fetch timeout operator-configurable via a new --p2p-block-sync-timeout flag and DEFRA_NET_P2PBLOCKSYNCTIMEOUT env var, and raise its default from 5s to 30s so deployments talking to a real SourceHub chain do not time out on the per-block access checks. Also add a per-request BlockSyncTimeout option to SyncDocuments (threaded through the HTTP and CLI clients) so an individual sync can override the node default, surface a block-fetch timeout as a distinct error rather than a generic load failure, and log when a receiver drops a pushed block it has no access to (previously silent). Refs https://github.com/sourcenetwork/defradb/issues/4837 --- cli/config/config.go | 2 + cli/p2p_document_sync.go | 7 +++- cli/start.go | 8 ++++ client/options/p2p.go | 22 +++++++++++ .../cli/defradb_client_p2p_document_sync.md | 5 ++- docs/website/references/cli/defradb_start.md | 1 + docs/website/references/http/openapi.json | 36 ++++++++++-------- http/client_p2p.go | 27 +++++++++++--- http/handler_p2p.go | 24 ++++++------ http/openapi.go | 1 + internal/db/config.go | 2 +- internal/db/p2p.go | 5 +++ internal/db/p2p/errors.go | 12 ++++++ internal/db/p2p/errors_test.go | 34 +++++++++++++++++ internal/db/p2p/p2p.go | 11 +++++- internal/db/p2p/sync_dag.go | 29 ++++++++++++++- internal/db/p2p/sync_dag_test.go | 37 +++++++++++++++++++ node/node.go | 2 +- tests/clients/cli/wrapper.go | 3 ++ tests/integration/p2p.go | 3 ++ tests/integration/test_case.go | 6 +++ 21 files changed, 239 insertions(+), 38 deletions(-) create mode 100644 internal/db/p2p/errors_test.go create mode 100644 internal/db/p2p/sync_dag_test.go 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..a330bc3868 100644 --- a/internal/db/p2p/errors.go +++ b/internal/db/p2p/errors.go @@ -72,6 +72,11 @@ 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 distinguishes a per-block fetch that ran out of time from other + // load failures. It usually means the peer serving the block was too slow to respond within + // the block-sync timeout (for policy-gated collections, often because its access check is + // slower than the timeout), rather than a decode or storage failure. + ErrBlockSyncTimeout = errors.New("timeout while fetching linked block during DAG sync") ) func NewErrReplicatorCollections(inner error, kv ...errors.KV) error { @@ -162,6 +167,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/errors_test.go b/internal/db/p2p/errors_test.go new file mode 100644 index 0000000000..97892763a2 --- /dev/null +++ b/internal/db/p2p/errors_test.go @@ -0,0 +1,34 @@ +// 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" + + "github.com/stretchr/testify/assert" + + "github.com/sourcenetwork/defradb/errors" +) + +// A block-fetch timeout must be identifiable as such and must not be confused with the generic +// linked-block load failure, so callers and operators can tell "the peer was too slow" apart +// from a decode/storage error. +func TestBlockSyncTimeoutError_IsDistinctAndWrapsCause(t *testing.T) { + err := NewErrBlockSyncTimeout(context.DeadlineExceeded, "bafyLink") + + assert.True(t, errors.Is(err, ErrBlockSyncTimeout), "should be identifiable as a block-sync timeout") + assert.True(t, errors.Is(err, context.DeadlineExceeded), "should preserve the deadline-exceeded cause") + + generic := NewErrLoadLinkedBlock(context.DeadlineExceeded) + assert.False(t, errors.Is(generic, ErrBlockSyncTimeout), + "the generic load error must not masquerade as a block-sync timeout") +} diff --git a/internal/db/p2p/p2p.go b/internal/db/p2p/p2p.go index 95c69d33c0..54f760e0d2 100644 --- a/internal/db/p2p/p2p.go +++ b/internal/db/p2p/p2p.go @@ -652,7 +652,16 @@ func (p *P2P) processPushlogRequest( return err } if !mightHaveAccess { - // If we know we don't have access, we can skip the rest of the processing. + // This node does not (yet) have read access to the document, so the pushed block is + // dropped. On the subscription path this commonly means the access grant has not + // propagated yet; the block will be retried when the sender announces it again. Log + // it so a dropped push can be told apart from one that never arrived. + log.Info( + "Dropping pushed block: no read access to document (grant may not have propagated yet)", + corelog.String("DocID", req.DocID), + corelog.String("CID", headCID.String()), + corelog.String("CollectionID", req.CollectionID), + ) return nil } } 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/p2p.go b/tests/integration/p2p.go index ebc8800157..2e7d1ed702 100644 --- a/tests/integration/p2p.go +++ b/tests/integration/p2p.go @@ -263,6 +263,9 @@ func syncDocs(s *state.State, action SyncDocs) { if identOption.HasValue() { syncOpts.SetIdentity(identOption.Value()) } + if action.BlockSyncTimeout.HasValue() { + syncOpts.SetBlockSyncTimeout(action.BlockSyncTimeout.Value()) + } err := withRetryOnNode( node, diff --git a/tests/integration/test_case.go b/tests/integration/test_case.go index fc0422318d..324b5244d8 100644 --- a/tests/integration/test_case.go +++ b/tests/integration/test_case.go @@ -12,6 +12,8 @@ package tests import ( + "time" + "github.com/sourcenetwork/immutable" "github.com/sourcenetwork/defradb/client" @@ -715,6 +717,10 @@ type SyncDocs struct { // There must an item for each document in DocIDs. SourceNodes []int + // BlockSyncTimeout, when set, overrides the node's default per-block fetch timeout for this + // sync only. Used to exercise the per-request timeout path. + BlockSyncTimeout immutable.Option[time.Duration] + // Any error expected from the action. ExpectedError string } From cbdbb2f5624ff000502c83337f06a7e8a44a8d45 Mon Sep 17 00:00:00 2001 From: Islam Aleiv Date: Thu, 9 Jul 2026 14:50:12 +0200 Subject: [PATCH 2/7] test: Cover per-request block-sync timeout end-to-end Add an integration test that drives SyncDocuments with a per-request BlockSyncTimeout too small for any block fetch to complete (the document does not sync) and again with a generous one (it syncs), proving the option is honoured through to the per-block fetch. Also correct a stale comment about the default block-sync timeout now that it is 30s. --- tests/integration/db.go | 5 +- .../sync/documents_block_sync_timeout_test.go | 99 +++++++++++++++++++ 2 files changed, 102 insertions(+), 2 deletions(-) create mode 100644 tests/integration/net/sync/documents_block_sync_timeout_test.go 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 diff --git a/tests/integration/net/sync/documents_block_sync_timeout_test.go b/tests/integration/net/sync/documents_block_sync_timeout_test.go new file mode 100644 index 0000000000..1050ef7fb4 --- /dev/null +++ b/tests/integration/net/sync/documents_block_sync_timeout_test.go @@ -0,0 +1,99 @@ +// Copyright 2026 Democratized Data Foundation +// +// This file is part of the DefraDB test suite. +// +// The DefraDB test suite is licensed under either: +// +// (1) GNU Affero General Public License v3 +// (2) Business Source License 1.1 +// +// See tests/LICENSE for details. + +package sync + +import ( + "testing" + "time" + + "github.com/sourcenetwork/immutable" + + "github.com/sourcenetwork/defradb/tests/action" + testUtils "github.com/sourcenetwork/defradb/tests/integration" +) + +// A per-request block-sync timeout that is too small for any block fetch to complete must abort the +// DAG fetch (the receiver logs the distinct block-sync timeout error and the document does not +// materialise). A generous per-request timeout on the same setup syncs cleanly. Together these +// exercise the SyncDocuments BlockSyncTimeout option end-to-end: the option threads through to +// loadBlockLinks and actually bounds the per-block fetch. +func TestDocSync_PerRequestBlockSyncTimeout_BoundsBlockFetch(t *testing.T) { + test := testUtils.TestCase{ + Actions: []any{ + testUtils.RandomNetworkingConfig(), + testUtils.RandomNetworkingConfig(), + &action.AddCollection{ + SDL: ` + type Users { + Name: String + Age: Int + } + `, + }, + &action.AddDoc{ + NodeID: immutable.Some(0), + Doc: `{ + "Name": "John", + "Age": 21 + }`, + }, + testUtils.ConnectPeers{ + SourceNodeID: 0, + TargetNodeID: 1, + }, + // A one-nanosecond per-block budget cannot be met, so the block fetch times out and the + // document is not synced to node 1. + testUtils.SyncDocs{ + NodeID: 1, + CollectionID: 0, + DocIDs: []int{0}, + SourceNodes: []int{0}, + BlockSyncTimeout: immutable.Some(time.Nanosecond), + }, + &action.Request{ + NodeID: immutable.Some(1), + Request: `query { + Users { + Name + } + }`, + Results: map[string]any{ + "Users": []map[string]any{}, + }, + }, + // A generous per-block budget on the same document syncs it through. + testUtils.SyncDocs{ + NodeID: 1, + CollectionID: 0, + DocIDs: []int{0}, + SourceNodes: []int{0}, + BlockSyncTimeout: immutable.Some(30 * time.Second), + }, + testUtils.WaitForSync{}, + &action.Request{ + NodeID: immutable.Some(1), + Request: `query { + Users { + Name + } + }`, + Results: map[string]any{ + "Users": []map[string]any{ + {"Name": "John"}, + }, + }, + }, + }, + } + + testUtils.ExecuteTestCase(t, test) +} From ac34995e64cab6c2045a25cf2c43a451c9868387 Mon Sep 17 00:00:00 2001 From: Islam Aleiv Date: Thu, 9 Jul 2026 15:26:37 +0200 Subject: [PATCH 3/7] refactor: Don't log receiver-side pushlog drops The sender re-announces a document's head block on every update, so logging each access-denied drop would spam a node running under document ACP once per update for every not-yet-granted document. The drop is an expected, self-healing transient (the block re-arrives once the grant propagates), so leave it unlogged and document why. --- internal/db/p2p/errors.go | 6 +----- internal/db/p2p/errors_test.go | 34 ---------------------------------- internal/db/p2p/p2p.go | 12 ++++-------- 3 files changed, 5 insertions(+), 47 deletions(-) delete mode 100644 internal/db/p2p/errors_test.go diff --git a/internal/db/p2p/errors.go b/internal/db/p2p/errors.go index a330bc3868..e5f9a34856 100644 --- a/internal/db/p2p/errors.go +++ b/internal/db/p2p/errors.go @@ -72,11 +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 distinguishes a per-block fetch that ran out of time from other - // load failures. It usually means the peer serving the block was too slow to respond within - // the block-sync timeout (for policy-gated collections, often because its access check is - // slower than the timeout), rather than a decode or storage failure. - ErrBlockSyncTimeout = errors.New("timeout while fetching linked block during DAG sync") + ErrBlockSyncTimeout = errors.New("timeout while fetching linked block during DAG sync") ) func NewErrReplicatorCollections(inner error, kv ...errors.KV) error { diff --git a/internal/db/p2p/errors_test.go b/internal/db/p2p/errors_test.go deleted file mode 100644 index 97892763a2..0000000000 --- a/internal/db/p2p/errors_test.go +++ /dev/null @@ -1,34 +0,0 @@ -// 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" - - "github.com/stretchr/testify/assert" - - "github.com/sourcenetwork/defradb/errors" -) - -// A block-fetch timeout must be identifiable as such and must not be confused with the generic -// linked-block load failure, so callers and operators can tell "the peer was too slow" apart -// from a decode/storage error. -func TestBlockSyncTimeoutError_IsDistinctAndWrapsCause(t *testing.T) { - err := NewErrBlockSyncTimeout(context.DeadlineExceeded, "bafyLink") - - assert.True(t, errors.Is(err, ErrBlockSyncTimeout), "should be identifiable as a block-sync timeout") - assert.True(t, errors.Is(err, context.DeadlineExceeded), "should preserve the deadline-exceeded cause") - - generic := NewErrLoadLinkedBlock(context.DeadlineExceeded) - assert.False(t, errors.Is(generic, ErrBlockSyncTimeout), - "the generic load error must not masquerade as a block-sync timeout") -} diff --git a/internal/db/p2p/p2p.go b/internal/db/p2p/p2p.go index 54f760e0d2..e4c0eea9ba 100644 --- a/internal/db/p2p/p2p.go +++ b/internal/db/p2p/p2p.go @@ -654,14 +654,10 @@ func (p *P2P) processPushlogRequest( if !mightHaveAccess { // This node does not (yet) have read access to the document, so the pushed block is // dropped. On the subscription path this commonly means the access grant has not - // propagated yet; the block will be retried when the sender announces it again. Log - // it so a dropped push can be told apart from one that never arrived. - log.Info( - "Dropping pushed block: no read access to document (grant may not have propagated yet)", - corelog.String("DocID", req.DocID), - corelog.String("CID", headCID.String()), - corelog.String("CollectionID", req.CollectionID), - ) + // propagated yet; the block is retried when the sender next announces it. This is an + // expected, self-healing transient, so it is intentionally not logged: the sender + // re-announces on every update, and logging here would spam a node running under + // document ACP once per update per not-yet-granted document. return nil } } From b9c3ec6b9e23d3606a530c72b34bfd3c8095fbdc Mon Sep 17 00:00:00 2001 From: Islam Aleiv Date: Thu, 9 Jul 2026 15:28:25 +0200 Subject: [PATCH 4/7] commit --- internal/db/p2p/p2p.go | 6 ------ 1 file changed, 6 deletions(-) diff --git a/internal/db/p2p/p2p.go b/internal/db/p2p/p2p.go index e4c0eea9ba..a89450a90e 100644 --- a/internal/db/p2p/p2p.go +++ b/internal/db/p2p/p2p.go @@ -652,12 +652,6 @@ func (p *P2P) processPushlogRequest( return err } if !mightHaveAccess { - // This node does not (yet) have read access to the document, so the pushed block is - // dropped. On the subscription path this commonly means the access grant has not - // propagated yet; the block is retried when the sender next announces it. This is an - // expected, self-healing transient, so it is intentionally not logged: the sender - // re-announces on every update, and logging here would spam a node running under - // document ACP once per update per not-yet-granted document. return nil } } From eec0e4a301dcdea8d7b3c51aebed82f8db964b33 Mon Sep 17 00:00:00 2001 From: Islam Aleiv Date: Thu, 9 Jul 2026 15:28:59 +0200 Subject: [PATCH 5/7] Commit --- internal/db/p2p/p2p.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/db/p2p/p2p.go b/internal/db/p2p/p2p.go index a89450a90e..95c69d33c0 100644 --- a/internal/db/p2p/p2p.go +++ b/internal/db/p2p/p2p.go @@ -652,6 +652,7 @@ func (p *P2P) processPushlogRequest( return err } if !mightHaveAccess { + // If we know we don't have access, we can skip the rest of the processing. return nil } } From 1783a6fa645d398985ebd567681b317f12deed48 Mon Sep 17 00:00:00 2001 From: Islam Aleiv Date: Fri, 10 Jul 2026 15:16:10 +0200 Subject: [PATCH 6/7] commit --- .../net/sync/documents_block_sync_timeout_test.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/integration/net/sync/documents_block_sync_timeout_test.go b/tests/integration/net/sync/documents_block_sync_timeout_test.go index 1050ef7fb4..d04e000b20 100644 --- a/tests/integration/net/sync/documents_block_sync_timeout_test.go +++ b/tests/integration/net/sync/documents_block_sync_timeout_test.go @@ -21,11 +21,11 @@ import ( testUtils "github.com/sourcenetwork/defradb/tests/integration" ) -// A per-request block-sync timeout that is too small for any block fetch to complete must abort the -// DAG fetch (the receiver logs the distinct block-sync timeout error and the document does not -// materialise). A generous per-request timeout on the same setup syncs cleanly. Together these -// exercise the SyncDocuments BlockSyncTimeout option end-to-end: the option threads through to -// loadBlockLinks and actually bounds the per-block fetch. +// A per-request block-sync timeout that is too small for any block fetch to complete aborts the +// DAG fetch, so the document does not materialise on the receiver. A generous per-request timeout +// on the same setup syncs it cleanly. Together these exercise the SyncDocuments BlockSyncTimeout +// option end-to-end: the option threads through to loadBlockLinks and actually bounds the +// per-block fetch (the generous case is the control that rules out an unrelated sync failure). func TestDocSync_PerRequestBlockSyncTimeout_BoundsBlockFetch(t *testing.T) { test := testUtils.TestCase{ Actions: []any{ From e8d7200ad4fb655f366edc8037b888f5a75afbb1 Mon Sep 17 00:00:00 2001 From: Islam Aleiv Date: Fri, 10 Jul 2026 15:48:26 +0200 Subject: [PATCH 7/7] test: Remove flaky per-request block-sync timeout test The test assumed a tiny per-block timeout would stop a document from syncing, but the per-block timeout only applies to a document's linked blocks; a freshly created single-block document has none, so it synced regardless and the negative assertion failed on the C-bindings client. The per-request timeout resolution is already covered by a deterministic unit test, so drop the integration test and the SyncDocs framework field that existed only for it. --- .../sync/documents_block_sync_timeout_test.go | 99 ------------------- tests/integration/p2p.go | 3 - tests/integration/test_case.go | 6 -- 3 files changed, 108 deletions(-) delete mode 100644 tests/integration/net/sync/documents_block_sync_timeout_test.go diff --git a/tests/integration/net/sync/documents_block_sync_timeout_test.go b/tests/integration/net/sync/documents_block_sync_timeout_test.go deleted file mode 100644 index d04e000b20..0000000000 --- a/tests/integration/net/sync/documents_block_sync_timeout_test.go +++ /dev/null @@ -1,99 +0,0 @@ -// Copyright 2026 Democratized Data Foundation -// -// This file is part of the DefraDB test suite. -// -// The DefraDB test suite is licensed under either: -// -// (1) GNU Affero General Public License v3 -// (2) Business Source License 1.1 -// -// See tests/LICENSE for details. - -package sync - -import ( - "testing" - "time" - - "github.com/sourcenetwork/immutable" - - "github.com/sourcenetwork/defradb/tests/action" - testUtils "github.com/sourcenetwork/defradb/tests/integration" -) - -// A per-request block-sync timeout that is too small for any block fetch to complete aborts the -// DAG fetch, so the document does not materialise on the receiver. A generous per-request timeout -// on the same setup syncs it cleanly. Together these exercise the SyncDocuments BlockSyncTimeout -// option end-to-end: the option threads through to loadBlockLinks and actually bounds the -// per-block fetch (the generous case is the control that rules out an unrelated sync failure). -func TestDocSync_PerRequestBlockSyncTimeout_BoundsBlockFetch(t *testing.T) { - test := testUtils.TestCase{ - Actions: []any{ - testUtils.RandomNetworkingConfig(), - testUtils.RandomNetworkingConfig(), - &action.AddCollection{ - SDL: ` - type Users { - Name: String - Age: Int - } - `, - }, - &action.AddDoc{ - NodeID: immutable.Some(0), - Doc: `{ - "Name": "John", - "Age": 21 - }`, - }, - testUtils.ConnectPeers{ - SourceNodeID: 0, - TargetNodeID: 1, - }, - // A one-nanosecond per-block budget cannot be met, so the block fetch times out and the - // document is not synced to node 1. - testUtils.SyncDocs{ - NodeID: 1, - CollectionID: 0, - DocIDs: []int{0}, - SourceNodes: []int{0}, - BlockSyncTimeout: immutable.Some(time.Nanosecond), - }, - &action.Request{ - NodeID: immutable.Some(1), - Request: `query { - Users { - Name - } - }`, - Results: map[string]any{ - "Users": []map[string]any{}, - }, - }, - // A generous per-block budget on the same document syncs it through. - testUtils.SyncDocs{ - NodeID: 1, - CollectionID: 0, - DocIDs: []int{0}, - SourceNodes: []int{0}, - BlockSyncTimeout: immutable.Some(30 * time.Second), - }, - testUtils.WaitForSync{}, - &action.Request{ - NodeID: immutable.Some(1), - Request: `query { - Users { - Name - } - }`, - Results: map[string]any{ - "Users": []map[string]any{ - {"Name": "John"}, - }, - }, - }, - }, - } - - testUtils.ExecuteTestCase(t, test) -} diff --git a/tests/integration/p2p.go b/tests/integration/p2p.go index 2e7d1ed702..ebc8800157 100644 --- a/tests/integration/p2p.go +++ b/tests/integration/p2p.go @@ -263,9 +263,6 @@ func syncDocs(s *state.State, action SyncDocs) { if identOption.HasValue() { syncOpts.SetIdentity(identOption.Value()) } - if action.BlockSyncTimeout.HasValue() { - syncOpts.SetBlockSyncTimeout(action.BlockSyncTimeout.Value()) - } err := withRetryOnNode( node, diff --git a/tests/integration/test_case.go b/tests/integration/test_case.go index 324b5244d8..fc0422318d 100644 --- a/tests/integration/test_case.go +++ b/tests/integration/test_case.go @@ -12,8 +12,6 @@ package tests import ( - "time" - "github.com/sourcenetwork/immutable" "github.com/sourcenetwork/defradb/client" @@ -717,10 +715,6 @@ type SyncDocs struct { // There must an item for each document in DocIDs. SourceNodes []int - // BlockSyncTimeout, when set, overrides the node's default per-block fetch timeout for this - // sync only. Used to exercise the per-request timeout path. - BlockSyncTimeout immutable.Option[time.Duration] - // Any error expected from the action. ExpectedError string }