NAC gate SyncBranchableCollection

AndrewSisley · AndrewSisley · commit de498d573eb1 · 2026-02-13T13:01:04.000-05:00
diff --git a/acp/types/types.go b/acp/types/types.go
@@ -110,6 +110,7 @@ const (
 	NodeP2PDocumentDeletePerm
 	NodeP2PDocumentListPerm
 	NodeP2PSyncCollectionVersionsPerm
+	NodeP2PSyncBranchableCollectionPerm
 	NodeSignatureVerifyPerm
 	NodeLensCreatePerm
 	NodeLensListPerm
@@ -157,6 +158,7 @@ var RequiredResourcePermissionsForNode = []string{
 	"p2p-document-delete",
 	"p2p-document-list",
 	"p2p-sync-collection-versions",
+	"p2p-sync-branchable-collection",
 	"signature-verify",
 	"lens-create",
 	"lens-list",
@@ -251,6 +253,8 @@ resources:
     expr: admin
   - name: p2p-sync-collection-versions
     expr: admin
+  - name: p2p-sync-branchable-collection
+    expr: admin
 
   - name: signature-verify
     expr: admin
diff --git a/cbindings/wrapper.go b/cbindings/wrapper.go
@@ -376,7 +376,12 @@ func (w *CWrapper) SyncCollectionVersions(
 	return nil
 }
 
-func (w *CWrapper) SyncBranchableCollection(ctx context.Context, collectionID string) error {
+func (w *CWrapper) SyncBranchableCollection(
+	ctx context.Context,
+	collectionID string,
+	opts ...options.Lister[options.SyncBranchableCollectionOptions],
+) error {
+	opt := utils.NewOptions(opts...)
 	cCollectionID := C.CString(collectionID)
 	defer C.free(unsafe.Pointer(cCollectionID))
 
@@ -388,8 +393,11 @@ func (w *CWrapper) SyncBranchableCollection(ctx context.Context, collectionID st
 	cTimerStr := C.CString(timerStr)
 	defer C.free(unsafe.Pointer(cTimerStr))
 
+	cIdentity := optionToUintptr(opt.GetIdentity())
+	defer C.IdentityFree(cIdentity)
+
 	res := ConvertAndFreeCResult(
-		C.P2PbranchableCollectionSync(C.uintptr_t(w.handle), cCollectionID, cTimerStr, C.uintptr_t(0)))
+		C.P2PbranchableCollectionSync(C.uintptr_t(w.handle), cCollectionID, cTimerStr, cIdentity))
 
 	if res.Status != 0 {
 		return errors.New(res.Error)
diff --git a/client/mocks/txn.go b/client/mocks/txn.go
diff --git a/client/options/p2p.go b/client/options/p2p.go
@@ -360,6 +360,35 @@ func (b *SyncCollectionVersionsOptionsBuilder) SetIdentity(id identity.Identity)
 	return b
 }
 
+// SyncBranchableCollectionOptions contains options for SyncBranchableCollection operation.
+type SyncBranchableCollectionOptions struct {
+	// Identity is the identity of the actor performing the operation.
+	Identity immutable.Option[identity.Identity]
+}
+
+// GetIdentity returns the identity for the operation.
+func (o *SyncBranchableCollectionOptions) GetIdentity() immutable.Option[identity.Identity] {
+	return o.Identity
+}
+
+// SyncBranchableCollectionOptionsBuilder is a builder for SyncBranchableCollectionOptions.
+type SyncBranchableCollectionOptionsBuilder struct {
+	enumerableBuilder[SyncBranchableCollectionOptions]
+}
+
+// SyncBranchableCollection creates a new SyncBranchableCollectionOptionsBuilder instance.
+func SyncBranchableCollection() *SyncBranchableCollectionOptionsBuilder {
+	return &SyncBranchableCollectionOptionsBuilder{}
+}
+
+// SetIdentity sets the identity for the operation.
+func (b *SyncBranchableCollectionOptionsBuilder) SetIdentity(id identity.Identity) *SyncBranchableCollectionOptionsBuilder {
+	b.append(func(opts *SyncBranchableCollectionOptions) {
+		opts.Identity = immutable.Some(id)
+	})
+	return b
+}
+
 // DeleteP2PDocumentsOptions contains options for RemoveP2PDocuments operation.
 type DeleteP2PDocumentsOptions struct {
 	// Identity is the identity of the actor performing the operation.
diff --git a/client/p2p.go b/client/p2p.go
@@ -123,7 +123,11 @@ type P2P interface {
 	// for branchable collections (collections marked with @branchable directive).
 	// It doesn't automatically subscribe to the collection for future updates.
 	// context.WithTimeout can be used to set a timeout for the operation.
-	SyncBranchableCollection(ctx context.Context, collectionID string) error
+	SyncBranchableCollection(
+		ctx context.Context,
+		collectionID string,
+		opts ...options.Lister[options.SyncBranchableCollectionOptions],
+	) error
 }
 
 type StreamHandler = func(stream io.Reader, peerID string)
diff --git a/http/client_p2p.go b/http/client_p2p.go
@@ -372,7 +372,12 @@ func (c *Client) SyncCollectionVersions(
 	return err
 }
 
-func (c *Client) SyncBranchableCollection(ctx context.Context, collectionID string) error {
+func (c *Client) SyncBranchableCollection(
+	ctx context.Context,
+	collectionID string,
+	opts ...options.Lister[options.SyncBranchableCollectionOptions],
+) error {
+
 	methodURL := c.http.apiURL.JoinPath("p2p", "collections", "sync-branchable")
 
 	req := map[string]any{
@@ -394,6 +399,8 @@ func (c *Client) SyncBranchableCollection(ctx context.Context, collectionID stri
 	// This is necessary because the node handling this request will usually wait whole timeout
 	// duration as it might receive responses from multiple peers.
 	httpCtx := context.Background()
+	opt := utils.NewOptions(opts...)
+	ctx = identity.WithContext(ctx, opt.GetIdentity())
 	if hasDeadline {
 		var cancel context.CancelFunc
 		httpCtx, cancel = context.WithTimeout(httpCtx, time.Until(deadline)+500*time.Millisecond)
diff --git a/internal/db/p2p.go b/internal/db/p2p.go
@@ -382,9 +382,19 @@ func (db *DB) SyncCollectionVersions(
 // context.WithTimeout can be used to set a timeout for the operation.
 //
 // WARNING: This function does not respect transactions.
-func (db *DB) SyncBranchableCollection(ctx context.Context, collectionID string) error {
+func (db *DB) SyncBranchableCollection(
+	ctx context.Context,
+	collectionID string,
+	opts ...options.Lister[options.SyncBranchableCollectionOptions],
+) error {
+	opt := utils.NewOptions(opts...)
+
+	if err := db.checkNodeAccess(ctx, opt.Identity, acpTypes.NodeP2PSyncBranchableCollectionPerm); err != nil {
+		return err
+	}
+
 	if db.p2p == nil {
 		return ErrNoP2P
 	}
-	return db.p2p.SyncBranchableCollection(ctx, collectionID)
+	return db.p2p.SyncBranchableCollection(ctx, collectionID, opt)
 }
diff --git a/internal/db/p2p/p2p.go b/internal/db/p2p/p2p.go
@@ -347,10 +347,17 @@ func (p *P2P) hasAccess(ctx context.Context, pid string, c cid.Cid) bool {
 		return true
 	}
 
-	cols, err := p.db.GetCollections(
-		ctx,
-		options.GetCollections().SetVersionID(block.Delta.GetCollectionVersionID()),
-	)
+	ident, err := p.db.GetNodeIdentity(p.ctx)
+	if err != nil {
+		log.ErrorE("Failed to get node identity", err)
+		return false
+	}
+	getColOpts := options.GetCollections().SetCollectionID(block.Delta.GetCollectionVersionID())
+	if ident.HasValue() {
+		getColOpts = getColOpts.SetIdentity(identity.FromDID(ident.Value().DID))
+	}
+
+	cols, err := p.db.GetCollections(ctx, getColOpts)
 	if err != nil {
 		log.ErrorE("Failed to get collections", err)
 		return false
diff --git a/internal/db/p2p/sync_branchable_col.go b/internal/db/p2p/sync_branchable_col.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/sourcenetwork/corelog"
 
+	"github.com/sourcenetwork/defradb/acp/identity"
 	"github.com/sourcenetwork/defradb/client"
 	"github.com/sourcenetwork/defradb/client/options"
 	"github.com/sourcenetwork/defradb/errors"
@@ -50,11 +51,17 @@ type syncBranchableCollectionReply struct {
 //
 // This function call will block until there is a response for the collection.
 // It is the responsibility of the caller to set an appropriate timeout on the context.
-func (p *P2P) SyncBranchableCollection(ctx context.Context, collectionID string) error {
-	cols, err := p.db.GetCollections(
-		ctx,
-		options.GetCollections().SetCollectionID(collectionID),
-	)
+func (p *P2P) SyncBranchableCollection(
+	ctx context.Context,
+	collectionID string,
+	opts *options.SyncBranchableCollectionOptions,
+) error {
+	getColOpts := options.GetCollections().SetCollectionID(collectionID)
+	if opts.Identity.HasValue() {
+		getColOpts = getColOpts.SetIdentity(opts.Identity.Value())
+	}
+
+	cols, err := p.db.GetCollections(ctx, getColOpts)
 	if err != nil {
 		return err
 	}
@@ -262,10 +269,16 @@ func (p *P2P) syncBranchableCollectionMessageHandler(from string, topic string,
 
 // processSyncBranchableCollection processes a branchable collection sync request and returns all head CIDs.
 func (p *P2P) processSyncBranchableCollection(collectionID string) ([][]byte, error) {
-	cols, err := p.db.GetCollections(
-		p.ctx,
-		options.GetCollections().SetCollectionID(collectionID),
-	)
+	ident, err := p.db.GetNodeIdentity(p.ctx)
+	if err != nil {
+		return nil, err
+	}
+	getColOpts := options.GetCollections().SetCollectionID(collectionID)
+	if ident.HasValue() {
+		getColOpts = getColOpts.SetIdentity(identity.FromDID(ident.Value().DID))
+	}
+
+	cols, err := p.db.GetCollections(p.ctx, getColOpts)
 	if err != nil || len(cols) == 0 {
 		return nil, err
 	}
diff --git a/internal/db/txn.go b/internal/db/txn.go
@@ -428,7 +428,11 @@ func (txn *Txn) SyncCollectionVersions(
 	return txn.db.SyncCollectionVersions(ctx, versionIDs, opts...)
 }
 
-func (txn *Txn) SyncBranchableCollection(ctx context.Context, collectionID string) error {
+func (txn *Txn) SyncBranchableCollection(
+	ctx context.Context,
+	collectionID string,
+	opts ...options.Lister[options.SyncBranchableCollectionOptions],
+) error {
 	ctx = InitContext(ctx, txn)
-	return txn.db.SyncBranchableCollection(ctx, collectionID)
+	return txn.db.SyncBranchableCollection(ctx, collectionID, opts...)
 }
diff --git a/tests/action/sync_branchable_collection.go b/tests/action/sync_branchable_collection.go
@@ -14,6 +14,10 @@ import (
 	"context"
 	"errors"
 	"time"
+
+	"github.com/sourcenetwork/defradb/client/options"
+	"github.com/sourcenetwork/defradb/tests/state"
+	"github.com/sourcenetwork/immutable"
 )
 
 // SyncBranchableCollection is an action that syncs a branchable collection's DAG
@@ -27,6 +31,11 @@ type SyncBranchableCollection struct {
 	// and subscribed to the collection's pubsub topics.
 	NodeID int
 
+	// The identity of this request. Optional.
+	//
+	// If node acp is enabled, identity will be used to check if this operation can be performed.
+	Identity immutable.Option[state.Identity]
+
 	// CollectionID is the index of the collection to sync.
 	CollectionID int
 
@@ -54,8 +63,14 @@ func (a *SyncBranchableCollection) Execute() {
 		return
 	}
 
+	opts := options.SyncBranchableCollection()
+	identOption := getIdentityForRequestSpecificToNode(a.s, a.Identity, a.NodeID)
+	if identOption.HasValue() {
+		opts.SetIdentity(identOption.Value())
+	}
+
 	collection := nodeState.Collections[a.CollectionID]
-	err := nodeState.SyncBranchableCollection(ctx, collection.CollectionID())
+	err := nodeState.SyncBranchableCollection(ctx, collection.CollectionID(), opts)
 
 	expectedErrorRaised := assertError(a.s.T, err, a.ExpectedError)
 	assertExpectedErrorRaised(a.s.T, a.ExpectedError, expectedErrorRaised)
diff --git a/tests/clients/cli/wrapper.go b/tests/clients/cli/wrapper.go
@@ -321,14 +321,21 @@ func (w *Wrapper) SyncCollectionVersions(
 	return err
 }
 
-func (w *Wrapper) SyncBranchableCollection(ctx context.Context, collectionID string) error {
+func (w *Wrapper) SyncBranchableCollection(
+	ctx context.Context,
+	collectionID string,
+	opts ...options.Lister[options.SyncBranchableCollectionOptions],
+) error {
 	args := []string{"client", "p2p", "collection", "sync-branchable", collectionID}
 
 	deadline, hasDeadline := ctx.Deadline()
 	if hasDeadline {
 		args = append(args, "--timeout", time.Until(deadline).String())
 	}
 
+	opt := utils.NewOptions(opts...)
+	args = appendIdentityArg(args, opt.GetIdentity())
+
 	_, err := w.cmd.execute(ctx, args)
 	return err
 }
diff --git a/tests/clients/http/wrapper.go b/tests/clients/http/wrapper.go
@@ -167,8 +167,11 @@ func (w *Wrapper) SyncCollectionVersions(
 	return w.client.SyncCollectionVersions(ctx, versionIDs, opts...)
 }
 
-func (w *Wrapper) SyncBranchableCollection(ctx context.Context, collectionID string) error {
-	return w.client.SyncBranchableCollection(ctx, collectionID)
+func (w *Wrapper) SyncBranchableCollection(
+	ctx context.Context,
+	collectionID string,
+	opts ...options.Lister[options.SyncBranchableCollectionOptions]) error {
+	return w.client.SyncBranchableCollection(ctx, collectionID, opts...)
 }
 
 func (w *Wrapper) BasicImport(ctx context.Context, filepath string) error {
diff --git a/tests/clients/js/wrapper.go b/tests/clients/js/wrapper.go
@@ -154,7 +154,7 @@ func (w *Wrapper) SyncCollectionVersions(ctx context.Context, versionIDs []strin
 	panic("not implemented")
 }
 
-func (w *Wrapper) SyncBranchableCollection(ctx context.Context, collectionID string) error {
+func (w *Wrapper) SyncBranchableCollection(ctx context.Context, collectionID string, opts ...options.Lister[options.SyncBranchableCollectionOptions]) error {
 	panic("not implemented")
 }
 
diff --git a/tests/integration/acp/nac/p2p_sync_branchable_collection_test.go b/tests/integration/acp/nac/p2p_sync_branchable_collection_test.go
diff --git a/tests/integration/acp/nac/relation_admin/p2p_sync_branchable_collection_test.go b/tests/integration/acp/nac/relation_admin/p2p_sync_branchable_collection_test.go

Original file line number	Diff line number	Diff line change
`@@ -428,7 +428,11 @@ func (txn *Txn) SyncCollectionVersions(`
`428`	`428`	`return txn.db.SyncCollectionVersions(ctx, versionIDs, opts...)`
`429`	`429`	`}`
`430`	`430`
`431`		`-func (txn *Txn) SyncBranchableCollection(ctx context.Context, collectionID string) error {`
	`431`	`+func (txn *Txn) SyncBranchableCollection(`
	`432`	`+ ctx context.Context,`
	`433`	`+ collectionID string,`
	`434`	`+ opts ...options.Lister[options.SyncBranchableCollectionOptions],`
	`435`	`+) error {`
`432`	`436`	`ctx = InitContext(ctx, txn)`
`433`		`- return txn.db.SyncBranchableCollection(ctx, collectionID)`
	`437`	`+ return txn.db.SyncBranchableCollection(ctx, collectionID, opts...)`
`434`	`438`	`}`