diff --git a/internal/db/merge.go b/internal/db/merge.go index 47db8740b1..211267e002 100644 --- a/internal/db/merge.go +++ b/internal/db/merge.go @@ -28,6 +28,7 @@ import ( "github.com/sourcenetwork/defradb/event" "github.com/sourcenetwork/defradb/internal/core" coreblock "github.com/sourcenetwork/defradb/internal/core/block" + "github.com/sourcenetwork/defradb/internal/core/crdt" "github.com/sourcenetwork/defradb/internal/db/base" "github.com/sourcenetwork/defradb/internal/encryption" "github.com/sourcenetwork/defradb/internal/keys" @@ -35,30 +36,30 @@ import ( merklecrdt "github.com/sourcenetwork/defradb/internal/merkle/crdt" ) -func (db *db) executeMerge(ctx context.Context, dagMerge event.Merge) error { +func (db *db) executeMerge(ctx context.Context, col *collection, dagMerge event.Merge) error { ctx, txn, err := ensureContextTxn(ctx, db, false) if err != nil { return err } defer txn.Discard(ctx) - col, err := getCollectionFromRootSchema(ctx, db, dagMerge.SchemaRoot) - if err != nil { - return err - } - - docID, err := client.NewDocIDFromString(dagMerge.DocID) - if err != nil { - return err - } - dsKey := base.MakeDataStoreKeyWithCollectionAndDocID(col.Description(), docID.String()) - - mp, err := db.newMergeProcessor(txn, col, dsKey) - if err != nil { - return err + var mt mergeTarget + if dagMerge.DocID != "" { + mt, err = getHeadsAsMergeTarget(ctx, txn, keys.HeadstoreDocKey{ + DocID: dagMerge.DocID, + FieldID: core.COMPOSITE_NAMESPACE, + }) + if err != nil { + return err + } + } else { + mt, err = getHeadsAsMergeTarget(ctx, txn, keys.NewHeadstoreColKey(col.Description().RootID)) + if err != nil { + return err + } } - mt, err := getHeadsAsMergeTarget(ctx, txn, dsKey.WithFieldID(core.COMPOSITE_NAMESPACE)) + mp, err := db.newMergeProcessor(txn, col) if err != nil { return err } @@ -73,9 +74,15 @@ func (db *db) executeMerge(ctx context.Context, dagMerge event.Merge) error { return err } - err = syncIndexedDoc(ctx, docID, col) - if err != nil { - return err + for docID := range mp.docIDs { + docID, err := client.NewDocIDFromString(docID) + if err != nil { + return err + } + err = syncIndexedDoc(ctx, docID, col) + if err != nil { + return err + } } err = txn.Commit(ctx) @@ -94,39 +101,39 @@ func (db *db) executeMerge(ctx context.Context, dagMerge event.Merge) error { // mergeQueue is synchronization source to ensure that concurrent // document merges do not cause transaction conflicts. type mergeQueue struct { - docs map[string]chan struct{} + keys map[string]chan struct{} mutex sync.Mutex } func newMergeQueue() *mergeQueue { return &mergeQueue{ - docs: make(map[string]chan struct{}), + keys: make(map[string]chan struct{}), } } -// add adds a docID to the queue. If the docID is already in the queue, it will -// wait for the docID to be removed from the queue. For every add call, done must -// be called to remove the docID from the queue. Otherwise, subsequent add calls will +// add adds a key to the queue. If the key is already in the queue, it will +// wait for the key to be removed from the queue. For every add call, done must +// be called to remove the key from the queue. Otherwise, subsequent add calls will // block forever. -func (m *mergeQueue) add(docID string) { +func (m *mergeQueue) add(key string) { m.mutex.Lock() - done, ok := m.docs[docID] + done, ok := m.keys[key] if !ok { - m.docs[docID] = make(chan struct{}) + m.keys[key] = make(chan struct{}) } m.mutex.Unlock() if ok { <-done - m.add(docID) + m.add(key) } } -func (m *mergeQueue) done(docID string) { +func (m *mergeQueue) done(key string) { m.mutex.Lock() defer m.mutex.Unlock() - done, ok := m.docs[docID] + done, ok := m.keys[key] if ok { - delete(m.docs, docID) + delete(m.keys, key) close(done) } } @@ -135,9 +142,11 @@ type mergeProcessor struct { txn datastore.Txn blockLS linking.LinkSystem encBlockLS linking.LinkSystem - mCRDTs map[string]merklecrdt.MerkleCRDT col *collection - dsKey keys.DataStoreKey + + // docIDs contains all docIDs that have been merged so far by the mergeProcessor + docIDs map[string]struct{} + // composites is a list of composites that need to be merged. composites *list.List // missingEncryptionBlocks is a list of blocks that we failed to fetch @@ -149,7 +158,6 @@ type mergeProcessor struct { func (db *db) newMergeProcessor( txn datastore.Txn, col *collection, - dsKey keys.DataStoreKey, ) (*mergeProcessor, error) { blockLS := cidlink.DefaultLinkSystem() blockLS.SetReadStorage(txn.Blockstore().AsIPLDStorage()) @@ -161,9 +169,8 @@ func (db *db) newMergeProcessor( txn: txn, blockLS: blockLS, encBlockLS: encBlockLS, - mCRDTs: make(map[string]merklecrdt.MerkleCRDT), col: col, - dsKey: dsKey, + docIDs: make(map[string]struct{}), composites: list.New(), missingEncryptionBlocks: make(map[cidlink.Link]struct{}), availableEncryptionBlocks: make(map[cidlink.Link]*coreblock.Encryption), @@ -375,7 +382,7 @@ func (mp *mergeProcessor) processBlock( } if canRead { - crdt, err := mp.initCRDTForType(dagBlock.Delta.GetFieldName()) + crdt, err := mp.initCRDTForType(dagBlock.Delta) if err != nil { return err } @@ -435,50 +442,59 @@ func decryptBlock( return newBlock, nil } -func (mp *mergeProcessor) initCRDTForType(field string) (merklecrdt.MerkleCRDT, error) { - mcrdt, exists := mp.mCRDTs[field] - if exists { - return mcrdt, nil - } - +func (mp *mergeProcessor) initCRDTForType(crdt crdt.CRDT) (merklecrdt.MerkleCRDT, error) { schemaVersionKey := keys.CollectionSchemaVersionKey{ SchemaVersionID: mp.col.Schema().VersionID, CollectionID: mp.col.ID(), } - if field == "" { - mcrdt = merklecrdt.NewMerkleCompositeDAG( + switch { + case crdt.IsComposite(): + docID := string(crdt.GetDocID()) + mp.docIDs[docID] = struct{}{} + + return merklecrdt.NewMerkleCompositeDAG( mp.txn, schemaVersionKey, - mp.dsKey.WithFieldID(core.COMPOSITE_NAMESPACE), - ) - mp.mCRDTs[field] = mcrdt - return mcrdt, nil - } + base.MakeDataStoreKeyWithCollectionAndDocID(mp.col.Description(), docID).WithFieldID(core.COMPOSITE_NAMESPACE), + ), nil - fd, ok := mp.col.Definition().GetFieldByName(field) - if !ok { - // If the field is not part of the schema, we can safely ignore it. - return nil, nil + case crdt.IsCollection(): + return merklecrdt.NewMerkleCollection( + mp.txn, + schemaVersionKey, + keys.NewHeadstoreColKey(mp.col.Description().RootID), + ), nil + + default: + docID := string(crdt.GetDocID()) + mp.docIDs[docID] = struct{}{} + + field := crdt.GetFieldName() + fd, ok := mp.col.Definition().GetFieldByName(field) + if !ok { + // If the field is not part of the schema, we can safely ignore it. + return nil, nil + } + + return merklecrdt.FieldLevelCRDTWithStore( + mp.txn, + schemaVersionKey, + fd.Typ, + fd.Kind, + base.MakeDataStoreKeyWithCollectionAndDocID(mp.col.Description(), docID).WithFieldID(fd.ID.String()), + field, + ) } +} - mcrdt, err := merklecrdt.FieldLevelCRDTWithStore( - mp.txn, - schemaVersionKey, - fd.Typ, - fd.Kind, - mp.dsKey.WithFieldID(fd.ID.String()), - field, - ) +func getCollectionFromRootSchema(ctx context.Context, db *db, rootSchema string) (*collection, error) { + ctx, txn, err := ensureContextTxn(ctx, db, false) if err != nil { return nil, err } + defer txn.Discard(ctx) - mp.mCRDTs[field] = mcrdt - return mcrdt, nil -} - -func getCollectionFromRootSchema(ctx context.Context, db *db, rootSchema string) (*collection, error) { cols, err := db.getCollections( ctx, client.CollectionFetchOptions{ @@ -498,8 +514,8 @@ func getCollectionFromRootSchema(ctx context.Context, db *db, rootSchema string) // getHeadsAsMergeTarget retrieves the heads of the composite DAG for the given document // and returns them as a merge target. -func getHeadsAsMergeTarget(ctx context.Context, txn datastore.Txn, dsKey keys.DataStoreKey) (mergeTarget, error) { - cids, err := getHeads(ctx, txn, dsKey) +func getHeadsAsMergeTarget(ctx context.Context, txn datastore.Txn, key keys.HeadstoreKey) (mergeTarget, error) { + cids, err := getHeads(ctx, txn, key) if err != nil { return mergeTarget{}, err @@ -520,8 +536,8 @@ func getHeadsAsMergeTarget(ctx context.Context, txn datastore.Txn, dsKey keys.Da } // getHeads retrieves the heads associated with the given datastore key. -func getHeads(ctx context.Context, txn datastore.Txn, dsKey keys.DataStoreKey) ([]cid.Cid, error) { - headset := clock.NewHeadSet(txn.Headstore(), dsKey.ToHeadStoreKey()) +func getHeads(ctx context.Context, txn datastore.Txn, key keys.HeadstoreKey) ([]cid.Cid, error) { + headset := clock.NewHeadSet(txn.Headstore(), key) cids, _, err := headset.List(ctx) if err != nil { diff --git a/internal/db/merge_test.go b/internal/db/merge_test.go index f9478be536..ee170bfe54 100644 --- a/internal/db/merge_test.go +++ b/internal/db/merge_test.go @@ -58,7 +58,7 @@ func TestMerge_SingleBranch_NoError(t *testing.T) { compInfo2, err := d.generateCompositeUpdate(&lsys, map[string]any{"name": "Johny"}, compInfo) require.NoError(t, err) - err = db.executeMerge(ctx, event.Merge{ + err = db.executeMerge(ctx, col.(*collection), event.Merge{ DocID: docID.String(), Cid: compInfo2.link.Cid, SchemaRoot: col.SchemaRoot(), @@ -103,7 +103,7 @@ func TestMerge_DualBranch_NoError(t *testing.T) { compInfo2, err := d.generateCompositeUpdate(&lsys, map[string]any{"name": "Johny"}, compInfo) require.NoError(t, err) - err = db.executeMerge(ctx, event.Merge{ + err = db.executeMerge(ctx, col.(*collection), event.Merge{ DocID: docID.String(), Cid: compInfo2.link.Cid, SchemaRoot: col.SchemaRoot(), @@ -113,7 +113,7 @@ func TestMerge_DualBranch_NoError(t *testing.T) { compInfo3, err := d.generateCompositeUpdate(&lsys, map[string]any{"age": 30}, compInfo) require.NoError(t, err) - err = db.executeMerge(ctx, event.Merge{ + err = db.executeMerge(ctx, col.(*collection), event.Merge{ DocID: docID.String(), Cid: compInfo3.link.Cid, SchemaRoot: col.SchemaRoot(), @@ -161,7 +161,7 @@ func TestMerge_DualBranchWithOneIncomplete_CouldNotFindCID(t *testing.T) { compInfo2, err := d.generateCompositeUpdate(&lsys, map[string]any{"name": "Johny"}, compInfo) require.NoError(t, err) - err = db.executeMerge(ctx, event.Merge{ + err = db.executeMerge(ctx, col.(*collection), event.Merge{ DocID: docID.String(), Cid: compInfo2.link.Cid, SchemaRoot: col.SchemaRoot(), @@ -180,7 +180,7 @@ func TestMerge_DualBranchWithOneIncomplete_CouldNotFindCID(t *testing.T) { compInfo3, err := d.generateCompositeUpdate(&lsys, map[string]any{"name": "Johny"}, compInfoUnkown) require.NoError(t, err) - err = db.executeMerge(ctx, event.Merge{ + err = db.executeMerge(ctx, col.(*collection), event.Merge{ DocID: docID.String(), Cid: compInfo3.link.Cid, SchemaRoot: col.SchemaRoot(), @@ -304,15 +304,15 @@ func TestMergeQueue(t *testing.T) { go q.add(testDocID) // give time for the goroutine to block time.Sleep(10 * time.Millisecond) - require.Len(t, q.docs, 1) + require.Len(t, q.keys, 1) q.done(testDocID) // give time for the goroutine to add the docID time.Sleep(10 * time.Millisecond) q.mutex.Lock() - require.Len(t, q.docs, 1) + require.Len(t, q.keys, 1) q.mutex.Unlock() q.done(testDocID) q.mutex.Lock() - require.Len(t, q.docs, 0) + require.Len(t, q.keys, 0) q.mutex.Unlock() } diff --git a/internal/db/messages.go b/internal/db/messages.go index 51efba982e..2dcc175cfe 100644 --- a/internal/db/messages.go +++ b/internal/db/messages.go @@ -22,7 +22,9 @@ import ( ) func (db *db) handleMessages(ctx context.Context, sub *event.Subscription) { - queue := newMergeQueue() + docIdQueue := newMergeQueue() + schemaRootQueue := newMergeQueue() + // This is used to ensure we only trigger loadAndPublishP2PCollections and loadAndPublishReplicators // once per db instanciation. loadOnce := sync.Once{} @@ -37,17 +39,34 @@ func (db *db) handleMessages(ctx context.Context, sub *event.Subscription) { switch evt := msg.Data.(type) { case event.Merge: go func() { - // ensure only one merge per docID - queue.add(evt.DocID) - defer queue.done(evt.DocID) + col, err := getCollectionFromRootSchema(ctx, db, evt.SchemaRoot) + if err != nil { + log.ErrorContextE( + ctx, + "Failed to execute merge", + err, + corelog.Any("Event", evt)) + return + } + + if col.Description().IsBranchable { + // As collection commits link to document composite commits, all events + // recieved for branchable collections must be processed serially else + // they may otherwise cause a transaction conflict. + schemaRootQueue.add(evt.SchemaRoot) + defer schemaRootQueue.done(evt.SchemaRoot) + } else { + // ensure only one merge per docID + docIdQueue.add(evt.DocID) + defer docIdQueue.done(evt.DocID) + } // retry the merge process if a conflict occurs // // conficts occur when a user updates a document // while a merge is in progress. - var err error for i := 0; i < db.MaxTxnRetries(); i++ { - err = db.executeMerge(ctx, evt) + err = db.executeMerge(ctx, col, evt) if errors.Is(err, datastore.ErrTxnConflict) { continue // retry merge } diff --git a/net/peer.go b/net/peer.go index e4ebfe8573..d59d6fe150 100644 --- a/net/peer.go +++ b/net/peer.go @@ -255,9 +255,11 @@ func (p *Peer) handleMessageLoop() { } func (p *Peer) handleLog(evt event.Update) error { - _, err := client.NewDocIDFromString(evt.DocID) - if err != nil { - return NewErrFailedToGetDocID(err) + if evt.DocID != "" { + _, err := client.NewDocIDFromString(evt.DocID) + if err != nil { + return NewErrFailedToGetDocID(err) + } } // push to each peer (replicator) @@ -273,8 +275,10 @@ func (p *Peer) handleLog(evt event.Update) error { Block: evt.Block, } - if err := p.server.publishLog(p.ctx, evt.DocID, req); err != nil { - return NewErrPublishingToDocIDTopic(err, evt.Cid.String(), evt.DocID) + if evt.DocID != "" { + if err := p.server.publishLog(p.ctx, evt.DocID, req); err != nil { + return NewErrPublishingToDocIDTopic(err, evt.Cid.String(), evt.DocID) + } } if err := p.server.publishLog(p.ctx, evt.SchemaRoot, req); err != nil { diff --git a/net/server.go b/net/server.go index c83ba3f6be..28ac5761ce 100644 --- a/net/server.go +++ b/net/server.go @@ -110,9 +110,12 @@ func (s *server) PushLog(ctx context.Context, req *pushLogRequest) (*pushLogRepl if err != nil { return nil, err } - docID, err := client.NewDocIDFromString(req.DocID) - if err != nil { - return nil, err + + if req.DocID != "" { + _, err := client.NewDocIDFromString(req.DocID) + if err != nil { + return nil, err + } } byPeer, err := libpeer.Decode(req.Creator) if err != nil { @@ -126,11 +129,11 @@ func (s *server) PushLog(ctx context.Context, req *pushLogRequest) (*pushLogRepl log.InfoContext(ctx, "Received pushlog", corelog.Any("PeerID", pid.String()), corelog.Any("Creator", byPeer.String()), - corelog.Any("DocID", docID.String())) + corelog.Any("DocID", req.DocID)) log.InfoContext(ctx, "Starting DAG sync", corelog.Any("PeerID", pid.String()), - corelog.Any("DocID", docID.String())) + corelog.Any("DocID", req.DocID)) err = syncDAG(ctx, s.peer.bserv, block) if err != nil { @@ -139,19 +142,19 @@ func (s *server) PushLog(ctx context.Context, req *pushLogRequest) (*pushLogRepl log.InfoContext(ctx, "DAG sync complete", corelog.Any("PeerID", pid.String()), - corelog.Any("DocID", docID.String())) + corelog.Any("DocID", req.DocID)) // Once processed, subscribe to the DocID topic on the pubsub network unless we already // subscribed to the collection. - if !s.hasPubSubTopicAndSubscribed(req.SchemaRoot) { - err = s.addPubSubTopic(docID.String(), true, nil) + if !s.hasPubSubTopicAndSubscribed(req.SchemaRoot) && req.DocID != "" { + _, err = s.addPubSubTopic(req.DocID, true, nil) if err != nil { return nil, err } } s.peer.bus.Publish(event.NewMessage(event.MergeName, event.Merge{ - DocID: docID.String(), + DocID: req.DocID, ByPeer: byPeer, FromPeer: pid, Cid: headCID, @@ -172,9 +175,9 @@ func (s *server) GetHeadLog( // addPubSubTopic subscribes to a topic on the pubsub network // A custom message handler can be provided to handle incoming messages. If not provided, // the default message handler will be used. -func (s *server) addPubSubTopic(topic string, subscribe bool, handler rpc.MessageHandler) error { +func (s *server) addPubSubTopic(topic string, subscribe bool, handler rpc.MessageHandler) (pubsubTopic, error) { if s.peer.ps == nil { - return nil + return pubsubTopic{}, nil } log.InfoContext(s.peer.ctx, "Adding pubsub topic", @@ -188,16 +191,16 @@ func (s *server) addPubSubTopic(topic string, subscribe bool, handler rpc.Messag // we need to close the existing topic and create a new one. if !t.subscribed && subscribe { if err := t.Close(); err != nil { - return err + return pubsubTopic{}, err } } else { - return nil + return t, nil } } t, err := rpc.NewTopic(s.peer.ctx, s.peer.ps, s.peer.host.ID(), topic, subscribe) if err != nil { - return err + return pubsubTopic{}, err } if handler == nil { @@ -206,15 +209,17 @@ func (s *server) addPubSubTopic(topic string, subscribe bool, handler rpc.Messag t.SetEventHandler(s.pubSubEventHandler) t.SetMessageHandler(handler) - s.topics[topic] = pubsubTopic{ + pst := pubsubTopic{ Topic: t, subscribed: subscribe, } - return nil + s.topics[topic] = pst + return pst, nil } func (s *server) AddPubSubTopic(topicName string, handler rpc.MessageHandler) error { - return s.addPubSubTopic(topicName, true, handler) + _, err := s.addPubSubTopic(topicName, true, handler) + return err } // hasPubSubTopicAndSubscribed checks if we are subscribed to a topic. @@ -274,13 +279,22 @@ func (s *server) publishLog(ctx context.Context, topic string, req *pushLogReque s.mu.Unlock() if !ok { subscribe := topic != req.SchemaRoot && !s.hasPubSubTopicAndSubscribed(req.SchemaRoot) - err := s.addPubSubTopic(topic, subscribe, nil) + + _, err := s.addPubSubTopic(topic, subscribe, nil) if err != nil { return errors.Wrap(fmt.Sprintf("failed to created single use topic %s", topic), err) } return s.publishLog(ctx, topic, req) } + if topic == req.SchemaRoot && req.DocID == "" && !t.subscribed { // todo - document + var err error + t, err = s.addPubSubTopic(topic, true, nil) + if err != nil { + return errors.Wrap(fmt.Sprintf("failed to created single use topic %s", topic), err) + } + } + log.InfoContext(ctx, "Publish log", corelog.String("PeerID", s.peer.PeerID().String()), corelog.String("Topic", topic)) @@ -356,7 +370,7 @@ func peerIDFromContext(ctx context.Context) (libpeer.ID, error) { func (s *server) updatePubSubTopics(evt event.P2PTopic) { for _, topic := range evt.ToAdd { - err := s.addPubSubTopic(topic, true, nil) + _, err := s.addPubSubTopic(topic, true, nil) if err != nil { log.ErrorE("Failed to add pubsub topic.", err) } diff --git a/tests/integration/query/commits/branchables/peer_index_test.go b/tests/integration/query/commits/branchables/peer_index_test.go new file mode 100644 index 0000000000..ab03eb3c56 --- /dev/null +++ b/tests/integration/query/commits/branchables/peer_index_test.go @@ -0,0 +1,68 @@ +// Copyright 2024 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 branchables + +import ( + "testing" + + "github.com/sourcenetwork/immutable" + + testUtils "github.com/sourcenetwork/defradb/tests/integration" +) + +func TestQueryCommitsBranchables_SyncsIndexAcrossPeerConnection(t *testing.T) { + test := testUtils.TestCase{ + Actions: []any{ + testUtils.RandomNetworkingConfig(), + testUtils.RandomNetworkingConfig(), + testUtils.SchemaUpdate{ + Schema: ` + type Users @branchable { + name: String @index + } + `, + }, + testUtils.ConnectPeers{ + SourceNodeID: 1, + TargetNodeID: 0, + }, + testUtils.SubscribeToCollection{ + NodeID: 1, + CollectionIDs: []int{0}, + }, + testUtils.CreateDoc{ + NodeID: immutable.Some(0), + Doc: `{ + "name": "John" + }`, + }, + testUtils.WaitForSync{}, + testUtils.Request{ + // This query errors out if the document's index has not been correctly + // constructed + Request: `query { + Users (filter: {name: {_eq: "John"}}){ + name + } + }`, + Results: map[string]any{ + "Users": []map[string]any{ + { + "name": "John", + }, + }, + }, + }, + }, + } + + testUtils.ExecuteTestCase(t, test) +} diff --git a/tests/integration/query/commits/branchables/peer_test.go b/tests/integration/query/commits/branchables/peer_test.go index 81ff77a240..8ba6bf7294 100644 --- a/tests/integration/query/commits/branchables/peer_test.go +++ b/tests/integration/query/commits/branchables/peer_test.go @@ -18,8 +18,6 @@ import ( testUtils "github.com/sourcenetwork/defradb/tests/integration" ) -// TODO: This test documents an unimplemented feature. Tracked by: -// https://github.com/sourcenetwork/defradb/issues/3212 func TestQueryCommitsBranchables_SyncsAcrossPeerConnection(t *testing.T) { test := testUtils.TestCase{ Actions: []any{ @@ -50,76 +48,153 @@ func TestQueryCommitsBranchables_SyncsAcrossPeerConnection(t *testing.T) { }, testUtils.WaitForSync{}, testUtils.Request{ - NodeID: immutable.Some(0), Request: `query { - commits { - cid - links { + commits { cid + links { + cid + } } - } - }`, + }`, Results: map[string]any{ "commits": []map[string]any{ { - "cid": testUtils.NewUniqueCid("collection"), + "cid": testUtils.NewUniqueCid(0), "links": []map[string]any{ { - "cid": testUtils.NewUniqueCid("composite"), + "cid": testUtils.NewUniqueCid(1), }, }, }, { - "cid": testUtils.NewUniqueCid("age"), + "cid": testUtils.NewUniqueCid(2), "links": []map[string]any{}, }, { - "cid": testUtils.NewUniqueCid("name"), + "cid": testUtils.NewUniqueCid(3), "links": []map[string]any{}, }, { - "cid": testUtils.NewUniqueCid("composite"), + "cid": testUtils.NewUniqueCid(1), "links": []map[string]any{ { - "cid": testUtils.NewUniqueCid("age"), + "cid": testUtils.NewUniqueCid(2), }, { - "cid": testUtils.NewUniqueCid("name"), + "cid": testUtils.NewUniqueCid(3), }, }, }, }, }, }, + }, + } + + testUtils.ExecuteTestCase(t, test) +} + +func TestQueryCommitsBranchables_SyncsMultipleAcrossPeerConnection(t *testing.T) { + test := testUtils.TestCase{ + Actions: []any{ + testUtils.RandomNetworkingConfig(), + testUtils.RandomNetworkingConfig(), + testUtils.SchemaUpdate{ + Schema: ` + type Users @branchable { + name: String + age: Int + } + `, + }, + testUtils.ConnectPeers{ + SourceNodeID: 1, + TargetNodeID: 0, + }, + testUtils.SubscribeToCollection{ + NodeID: 1, + CollectionIDs: []int{0}, + }, + testUtils.CreateDoc{ + NodeID: immutable.Some(0), + Doc: `{ + "name": "John", + "age": 21 + }`, + }, + testUtils.CreateDoc{ + NodeID: immutable.Some(0), + Doc: `{ + "name": "Fred", + "age": 25 + }`, + }, + testUtils.WaitForSync{}, testUtils.Request{ - NodeID: immutable.Some(1), Request: `query { - commits { - cid - links { + commits { cid + links { + cid + } } - } - }`, + }`, Results: map[string]any{ "commits": []map[string]any{ - // Note: The collection commit has not synced. { - "cid": testUtils.NewUniqueCid("age"), + "cid": testUtils.NewUniqueCid("collection, doc2 create"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("collection, doc1 create"), + }, + { + "cid": testUtils.NewUniqueCid("doc2 create"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("collection, doc1 create"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("doc1 create"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("doc1 name"), + "links": []map[string]any{}, + }, + { + "cid": testUtils.NewUniqueCid("doc1 age"), + "links": []map[string]any{}, + }, + { + "cid": testUtils.NewUniqueCid("doc1 create"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("doc1 name"), + }, + { + "cid": testUtils.NewUniqueCid("doc1 age"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("doc2 name"), "links": []map[string]any{}, }, { - "cid": testUtils.NewUniqueCid("name"), + "cid": testUtils.NewUniqueCid("doc2 age"), "links": []map[string]any{}, }, { - "cid": testUtils.NewUniqueCid("composite"), + "cid": testUtils.NewUniqueCid("doc2 create"), "links": []map[string]any{ { - "cid": testUtils.NewUniqueCid("age"), + "cid": testUtils.NewUniqueCid("doc2 name"), }, { - "cid": testUtils.NewUniqueCid("name"), + "cid": testUtils.NewUniqueCid("doc2 age"), }, }, }, diff --git a/tests/integration/query/commits/branchables/peer_update_test.go b/tests/integration/query/commits/branchables/peer_update_test.go new file mode 100644 index 0000000000..e9e2c1b7d6 --- /dev/null +++ b/tests/integration/query/commits/branchables/peer_update_test.go @@ -0,0 +1,209 @@ +// Copyright 2024 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 branchables + +import ( + "testing" + + "github.com/sourcenetwork/immutable" + + testUtils "github.com/sourcenetwork/defradb/tests/integration" +) + +func TestQueryCommitsBranchables_HandlesConcurrentUpdatesAcrossPeerConnection(t *testing.T) { + test := testUtils.TestCase{ + Actions: []any{ + testUtils.RandomNetworkingConfig(), + testUtils.RandomNetworkingConfig(), + testUtils.SchemaUpdate{ + Schema: ` + type Users @branchable { + name: String + } + `, + }, + testUtils.CreateDoc{ + Doc: `{ + "name": "John" + }`, + }, + testUtils.UpdateDoc{ + NodeID: immutable.Some(0), + Doc: `{ + "name": "Fred" + }`, + }, + testUtils.UpdateDoc{ + NodeID: immutable.Some(1), + Doc: `{ + "name": "Shahzad" + }`, + }, + testUtils.ConnectPeers{ + SourceNodeID: 1, + TargetNodeID: 0, + }, + testUtils.SubscribeToCollection{ + NodeID: 1, + CollectionIDs: []int{0}, + }, + testUtils.WaitForSync{}, + testUtils.UpdateDoc{ + NodeID: immutable.Some(1), + Doc: `{ + "name": "Chris" + }`, + }, + testUtils.WaitForSync{}, + testUtils.Request{ + NodeID: immutable.Some(0), //todo! + Request: `query { + commits { + cid + links { + cid + } + } + }`, + Results: map[string]any{ + "commits": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("collection, update2"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("collection, node1 update1"), + }, + { + "cid": testUtils.NewUniqueCid("doc, update2"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("collection, node1 update1"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("collection, create"), + }, + { + "cid": testUtils.NewUniqueCid("doc, node1 update1"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("collection, create"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("doc, create"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("collection, node0 update1"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("collection, create"), + }, + { + "cid": testUtils.NewUniqueCid("doc, node0 update1"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("name, node0 update1"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("name, create"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("name, create"), + "links": []map[string]any{}, + }, + { + "cid": testUtils.NewUniqueCid("name, update2"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("name, node1 update1"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("name, node1 update1"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("name, create"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("doc, update2"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("doc, node1 update1"), + }, + { + "cid": testUtils.NewUniqueCid("name, update2"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("doc, node1 update1"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("doc, create"), + }, + { + "cid": testUtils.NewUniqueCid("name, node1 update1"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("doc, create"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("name, create"), + }, + }, + }, + { + "cid": testUtils.NewUniqueCid("doc, node0 update1"), + "links": []map[string]any{ + { + "cid": testUtils.NewUniqueCid("doc, create"), + }, + { + "cid": testUtils.NewUniqueCid("name, node0 update1"), + }, + }, + }, + }, + }, + }, + testUtils.Request{ + Request: `query { + Users { + name + } + }`, + Results: map[string]any{ + "Users": []map[string]any{ + { + "name": "Chris", + }, + }, + }, + }, + }, + } + + testUtils.ExecuteTestCase(t, test) +}