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..0be9def0ce 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,23 @@ 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 {
+		// If the push log request is scoped to the schema and not to a document, subscribe to the
+		// schema.
+		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 +371,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..3001a291b0
--- /dev/null
+++ b/tests/integration/query/commits/branchables/peer_update_test.go
@@ -0,0 +1,300 @@
+// 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.WaitForSync{},
+			testUtils.UpdateDoc{
+				NodeID: immutable.Some(1),
+				Doc: `{
+					"name":	"Chris"
+				}`,
+			},
+			testUtils.WaitForSync{},
+			testUtils.Request{
+				NodeID: immutable.Some(0),
+				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{
+				NodeID: immutable.Some(1),
+				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("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("name, create"),
+							"links": []map[string]any{},
+						},
+						{
+							"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"),
+								},
+							},
+						},
+					},
+				},
+			},
+			testUtils.Request{
+				Request: `query {
+					Users {
+						name
+					}
+				}`,
+				Results: map[string]any{
+					"Users": []map[string]any{
+						{
+							"name": "Chris",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	testUtils.ExecuteTestCase(t, test)
+}