chore: count and report merge outcomes

Author: mennatnaga

internal/db/db.go

@@ -125,6 +125,9 @@ type DB struct {
 	lockSet *lock.LockSet
 
 	collectionRepository *description.CollectionRepository
+
+	// stats are the merge-path counters reported on an interval by reportMergeStats.
+	stats *mergeStats
 }
 
 var _ client.TxnStore = (*DB)(nil)
@@ -176,6 +179,7 @@ func newDB(
 		p2pBlockSyncTimeout:     cfg.P2PBlockSyncTimeout,
 		lockSet:                 lockSet,
 		collectionRepository:    description.NewColCache(lockSet, datastore.NewUnsafeDatastore(rootstore)),
+		stats:                   &mergeStats{},
 	}
 
 	lensRuntime, err := newLensRuntime(LensRuntimeType(cfg.LensRuntime))
@@ -233,6 +237,8 @@ func newDB(
 		}
 	})
 
+	go db.reportMergeStats(db.ctx)
+
 	return db, nil
 }
 

internal/db/merge.go

@@ -15,6 +15,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"strings"
 	"sync"
 
 	"github.com/ipfs/go-cid"
@@ -76,6 +77,12 @@ func (db *DB) Merge(ctx context.Context, evt event.Merge) error {
 // several, so a bigger chunk sorts a bigger set more times.
 const mergeChunkSize = 8
 
+// Phases a merge chunk can fail in, reported on the retry-exhaustion log line.
+const (
+	phaseRead   = "read"
+	phaseCommit = "commit"
+)
+
 type mergeEntry struct {
 	evt event.Merge
 	col *collection
@@ -106,6 +113,7 @@ func (db *DB) MergeBatchWithTxn(ctx context.Context, merges []event.Merge) ([]bo
 		col, err := getCollectionFromCollectionID(ctx, db, evt.CollectionID)
 		if err != nil {
 			errs = append(errs, NewErrMergeEventDropped(err, evt.DocID, evt.Cid.String()))
+			db.stats.markDropped(dropCollection)
 			continue
 		}
 		entries = append(entries, mergeEntry{evt: evt, col: col, index: i})
@@ -178,6 +186,7 @@ func (db *DB) MergeBatchWithTxn(ctx context.Context, merges []event.Merge) ([]bo
 		for i := range chunk {
 			if err := db.mergeChunk(ctx, chunk[i:i+1]); err != nil {
 				errs = append(errs, NewErrMergeEventDropped(err, chunk[i].evt.DocID, chunk[i].evt.Cid.String()))
+				db.stats.markDropped(mergeDropReason(err))
 				continue
 			}
 			db.publishMergeComplete(chunk[i : i+1])
@@ -197,29 +206,50 @@ func (db *DB) txnAttempts() int {
 	return 1
 }
 
+// namedDocs renders the documents a transaction touched, as collection/docID. Badger
+// reports conflicts without naming the contended key, so this is the only lead available
+// for working out which documents contend with each other.
+func namedDocs(entries []mergeEntry) string {
+	docIDs := make([]string, len(entries))
+	for i, e := range entries {
+		docIDs[i] = e.col.Name() + "/" + e.evt.DocID
+	}
+	return strings.Join(docIDs, ",")
+}
+
 // mergeChunk merges every event of the chunk inside one transaction, retrying the
 // whole chunk on transaction conflict. Isolating a failing event is the caller's job.
 func (db *DB) mergeChunk(ctx context.Context, entries []mergeEntry) error {
-	// Held so that exhausting the retry budget can report the conflict that caused it.
+	// Held so that exhausting the retry budget can report the conflict that caused it and
+	// where the last one was raised.
 	var conflictErr error
+	var phase string
+	// Whether each event created its document, kept until the transaction commits so a
+	// retried attempt does not count its events twice.
+	creates := make([]bool, 0, len(entries))
 	for i := 0; i < db.txnAttempts(); i++ {
 		txn, err := db.NewTxn(false)
 		if err != nil {
 			return err
 		}
 		txnCtx := InitContext(ctx, txn)
 
+		creates = creates[:0]
 		var mergeErr error
 		for _, e := range entries {
-			if mergeErr = db.mergeInTxn(txnCtx, e.col, e.evt); mergeErr != nil {
+			isCreate, err := db.mergeInTxn(txnCtx, e.col, e.evt)
+			if err != nil {
+				mergeErr, phase = err, phaseRead
 				break
 			}
+			creates = append(creates, isCreate)
 		}
 
 		if mergeErr != nil {
 			txn.Discard()
 			if errors.Is(mergeErr, corekv.ErrTxnConflict) {
 				conflictErr = mergeErr
+				db.stats.chunkConflicts.Add(1)
 				continue
 			}
 			return mergeErr
@@ -229,15 +259,29 @@ func (db *DB) mergeChunk(ctx context.Context, entries []mergeEntry) error {
 			txn.Discard()
 			if errors.Is(err, corekv.ErrTxnConflict) {
 				conflictErr = err
+				db.stats.chunkConflicts.Add(1)
+				phase = phaseCommit
 				continue
 			}
 			return err
 		}
 
+		for _, isCreate := range creates {
+			db.stats.markCreateOrUpdate(isCreate)
+		}
 		return nil
 	}
 
-	// Nothing was committed, so callers must not treat the events as merged.
+	// The chunk used its whole retry budget without committing. The caller then re-runs it
+	// one event at a time, so this counts conflict pressure rather than loss. What was
+	// actually lost is named in the caller's error.
+	db.stats.markExhausted()
+
+	log.InfoContext(ctx, "merge chunk exhausted its retries",
+		corelog.Int("attempts", db.txnAttempts()),
+		corelog.String("phase", phase),
+		corelog.String("docIDs", namedDocs(entries)),
+	)
 	return client.NewErrMaxTxnRetries(conflictErr)
 }
 
@@ -254,13 +298,15 @@ func (db *DB) executeMerge(ctx context.Context, col *collection, dagMerge event.
 	}
 	defer txn.Discard()
 
-	if err := db.mergeInTxn(ctx, col, dagMerge); err != nil {
+	isCreate, err := db.mergeInTxn(ctx, col, dagMerge)
+	if err != nil {
 		return err
 	}
 
 	if err := txn.Commit(); err != nil {
 		return err
 	}
+	db.stats.markCreateOrUpdate(isCreate)
 
 	// send a complete event so we can track merges in the integration tests
 	db.events.Publish(event.NewMessage(event.MergeCompleteName, event.MergeComplete{Merge: dagMerge}))
@@ -269,40 +315,47 @@ func (db *DB) executeMerge(ctx context.Context, col *collection, dagMerge event.
 
 // mergeInTxn executes the merge logic for a single event using the transaction already
 // present on ctx.  It does not commit; the caller is responsible for committing.
-func (db *DB) mergeInTxn(ctx context.Context, col *collection, dagMerge event.Merge) error {
+//
+// Reports whether the event created the document rather than updating one already held,
+// which the caller counts once the transaction commits.
+func (db *DB) mergeInTxn(ctx context.Context, col *collection, dagMerge event.Merge) (bool, error) {
 	key, exists, err := getDocHeadstoreKey(ctx, col, dagMerge.DocID)
 	if err != nil {
-		return err
+		return false, err
 	}
 
 	mt := newMergeTarget()
 	if exists {
 		mt, err = getHeadsAsMergeTarget(ctx, key)
 		if err != nil {
-			return NewErrGetMergeTargetHeads(err, dagMerge.DocID, string(key.Bytes()))
+			return false, NewErrGetMergeTargetHeads(err, dagMerge.DocID, string(key.Bytes()))
 		}
 	}
 
-	mp, err := db.newMergeProcessor(ctx, col, len(mt.heads) == 0)
+	// No local heads means the merge is creating the document rather than updating one
+	// that already exists here.
+	newDocCreateMode := len(mt.heads) == 0
+
+	mp, err := db.newMergeProcessor(ctx, col, newDocCreateMode)
 	if err != nil {
-		return err
+		return false, err
 	}
 
 	if err = mp.loadComposites(ctx, dagMerge.Cid, mt); err != nil {
-		return NewErrLoadComposites(err, dagMerge.Cid.String(), dagMerge.DocID)
+		return false, NewErrLoadComposites(err, dagMerge.Cid.String(), dagMerge.DocID)
 	}
 
 	if err = mp.mergeComposites(ctx); err != nil {
-		return NewErrMergeComposites(err, dagMerge.DocID)
+		return false, NewErrMergeComposites(err, dagMerge.DocID)
 	}
 
 	for docID, oldDoc := range mp.docIDs {
 		if err = syncIndexedDoc(ctx, docID, mp.col, oldDoc); err != nil {
-			return NewErrSyncIndexedDoc(err, docID.String())
+			return false, NewErrSyncIndexedDoc(err, docID.String())
 		}
 	}
 
-	return nil
+	return newDocCreateMode, nil
 }
 
 const maxConcurrentMerges = 32

internal/db/stats.go

@@ -0,0 +1,170 @@
+// 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 db
+
+import (
+	"context"
+	"log/slog"
+	"maps"
+	"slices"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	ipld "github.com/ipfs/go-ipld-format"
+
+	"github.com/sourcenetwork/corelog"
+
+	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/defradb/errors"
+)
+
+// mergeStatsInterval matches the p2p reporter so the two lines of a single interval line
+// up in the log.
+const mergeStatsInterval = 30 * time.Second
+
+// mergeStats are the merge-path counters reported once per interval and reset on report,
+// so each line carries the rate for that interval rather than a running total.
+//
+// Counting is an atomic add on paths that already do storage work, so it is not a cost
+// worth gating. Nothing here builds a string or allocates until the reporter runs.
+type mergeStats struct {
+	// creates and updates split merges by whether the document already had heads locally.
+	// A merge with no local heads is creating the document.
+	creates atomic.Int64
+	updates atomic.Int64
+
+	// chunkConflicts counts merge attempts abandoned for a transaction conflict, whether
+	// or not a later attempt succeeded. chunkExhausted counts the chunks that ran out of
+	// attempts, which is where a conflict becomes a dropped document.
+	chunkConflicts atomic.Int64
+	chunkExhausted atomic.Int64
+
+	// dropReasons counts dropped events by cause. A dropped event is a document this node
+	// did not store, and the causes need different responses, so the total on its own does
+	// not say what to do.
+	dropMu      sync.Mutex
+	dropReasons map[string]int64
+}
+
+// Causes a merge event can be dropped for. An unrecognised cause counts as dropOther, so
+// that staying non-zero means this list has fallen behind the code.
+const (
+	dropMissingBlock   = "missingBlock"
+	dropUniqueIndex    = "uniqueIndex"
+	dropRetryExhausted = "retryExhausted"
+	dropCollection     = "collectionNotFound"
+	dropOther          = "other"
+)
+
+// mergeDropReason names why an event was dropped: the sender could not supply the DAG,
+// two documents claim one indexed value, or the write kept losing to a concurrent one.
+func mergeDropReason(err error) string {
+	switch {
+	case errors.Is(err, ipld.ErrNotFound{}):
+		return dropMissingBlock
+	case errors.Is(err, errors.New(errCanNotIndexNonUniqueFields)):
+		return dropUniqueIndex
+	case errors.Is(err, client.NewErrMaxTxnRetries(nil)):
+		return dropRetryExhausted
+	default:
+		return dropOther
+	}
+}
+
+// markDropped records an event that did not merge, under the given cause.
+func (s *mergeStats) markDropped(reason string) {
+	if s == nil {
+		return
+	}
+	s.dropMu.Lock()
+	defer s.dropMu.Unlock()
+	if s.dropReasons == nil {
+		s.dropReasons = make(map[string]int64)
+	}
+	s.dropReasons[reason]++
+}
+
+// drainDropReasons returns the causes counted since the last call and resets them.
+func (s *mergeStats) drainDropReasons() []slog.Attr {
+	s.dropMu.Lock()
+	defer s.dropMu.Unlock()
+	if len(s.dropReasons) == 0 {
+		return nil
+	}
+	reasons := slices.Sorted(maps.Keys(s.dropReasons))
+	attrs := make([]slog.Attr, 0, len(reasons))
+	for _, reason := range reasons {
+		attrs = append(attrs, corelog.Int64(reason, s.dropReasons[reason]))
+	}
+	clear(s.dropReasons)
+	return attrs
+}
+
+// markCreateOrUpdate records whether a merge is creating the document or updating one that
+// already exists locally.
+func (s *mergeStats) markCreateOrUpdate(isCreate bool) {
+	if s == nil {
+		return
+	}
+	if isCreate {
+		s.creates.Add(1)
+		return
+	}
+	s.updates.Add(1)
+}
+
+// markExhausted records a chunk that ran out of attempts.
+func (s *mergeStats) markExhausted() {
+	if s == nil {
+		return
+	}
+	s.chunkExhausted.Add(1)
+}
+
+// reportMergeStats logs the merge counters once per interval until the database context is
+// cancelled. Rates are reported per interval rather than per event, which keeps the merge
+// path quiet under load where per-event logging would dominate the output.
+func (db *DB) reportMergeStats(ctx context.Context) {
+	ticker := time.NewTicker(mergeStatsInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			db.stats.report()
+		}
+	}
+}
+
+func (s *mergeStats) report() {
+	creates := s.creates.Swap(0)
+	updates := s.updates.Swap(0)
+	conflicts := s.chunkConflicts.Swap(0)
+	exhausted := s.chunkExhausted.Swap(0)
+
+	// Nothing to report on an idle database, or one doing only local writes.
+	if creates != 0 || updates != 0 || conflicts != 0 || exhausted != 0 {
+		log.Info("merge stats",
+			corelog.Int64("creates", creates),
+			corelog.Int64("updates", updates),
+			corelog.Int64("chunkConflicts", conflicts),
+			corelog.Int64("exhausted", exhausted),
+		)
+	}
+
+	// Its own line, so an interval with no drops stays quiet and the line lists only the
+	// causes that occurred.
+	if drops := s.drainDropReasons(); len(drops) > 0 {
+		log.Error("merge drops", drops...)
+	}
+}