chore: merge, ingest and store telemetry

client/errors.go

@@ -107,6 +107,7 @@ var (
 	ErrNullValueForNonNillableField          = errors.New(errNullValueForNonNillableField)
 	ErrMissingRequiredField                  = errors.New(errMissingRequiredField)
 	ErrTransactionNotFound                   = errors.New(errTransactionNotFound)
+	ErrMaxTxnRetries                         = errors.New(errMaxTxnRetries)
 )
 
 // NewErrFieldNotExist returns an error indicating that the given field does not exist.

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/index.go

@@ -14,6 +14,7 @@ import (
 	"context"
 
 	"github.com/sourcenetwork/corekv"
+	"github.com/sourcenetwork/corelog"
 
 	"github.com/sourcenetwork/defradb/client"
 	"github.com/sourcenetwork/defradb/errors"
@@ -438,6 +439,8 @@ func saveUniqueKey(
 	}
 
 	if len(val) != 0 {
+		// This read puts the unique key in the transaction's read set, so two transactions
+		// writing the same index value conflict at commit rather than at this check.
 		existing, err := txn.Datastore().Get(ctx, &key)
 		if err != nil && !errors.Is(err, corekv.ErrNotFound) {
 			return NewErrCheckUniqueIndexConstraint(err)
@@ -446,6 +449,16 @@ func saveUniqueKey(
 			if tolerateSameDoc && string(existing) == string(val) {
 				return nil
 			}
+			// The holder stays on this node: the error is rendered to whoever asked for the
+			// write, and on the merge path that is the peer that pushed it. An empty holder
+			// means the entry could not be resolved to a document.
+			var heldBy string
+			if shortID, decodeErr := keys.DecodeDocShortID(existing); decodeErr == nil {
+				heldBy, _, _ = id.GetDocID(ctx, shortID)
+			}
+			log.InfoContext(ctx, "unique index violation",
+				corelog.String("docID", doc.ID().String()),
+				corelog.String("heldBy", heldBy))
 			return newUniqueIndexError(doc, fieldsDescs)
 		}
 	}

internal/db/index_backfill_test.go

@@ -370,3 +370,39 @@ func TestBackfillBatchTxn_ConflictsWhenReadDocIsModified(t *testing.T) {
 	require.True(t, errors.Is(commitErr, corekv.ErrTxnConflict),
 		"expected ErrTxnConflict but got: %v", commitErr)
 }
+
+// The document already holding a contested value is one the writer never named and, under
+// document access control, may not be allowed to read. The error goes back to the writer,
+// and on the merge path to the peer that pushed the log, so it names only the writer.
+func TestSaveUniqueKey_DoesNotNameTheDocumentHoldingTheValue(t *testing.T) {
+	ctx := context.Background()
+
+	db, err := newBadgerDB(ctx)
+	require.NoError(t, err)
+	t.Cleanup(func() { db.Close() })
+
+	_, err = db.AddCollection(ctx, userSchema)
+	require.NoError(t, err)
+	col, err := db.GetCollectionByName(ctx, "User")
+	require.NoError(t, err)
+
+	_, err = col.NewIndex(ctx, client.NewIndexRequest{
+		Fields: []client.IndexedFieldDescription{{Name: "name"}},
+		Unique: true,
+	})
+	require.NoError(t, err)
+
+	holder, err := client.NewDocFromJSON(ctx, []byte(`{"name":"alice","age":1}`), col.Version())
+	require.NoError(t, err)
+	require.NoError(t, col.AddDocument(ctx, holder))
+
+	// Same indexed value, different age, so a different document that cannot have the slot.
+	duplicate, err := client.NewDocFromJSON(ctx, []byte(`{"name":"alice","age":2}`), col.Version())
+	require.NoError(t, err)
+	require.NotEqual(t, holder.ID().String(), duplicate.ID().String())
+
+	err = col.AddDocument(ctx, duplicate)
+	require.ErrorIs(t, err, ErrCanNotIndexNonUniqueFields)
+	require.Contains(t, err.Error(), duplicate.ID().String(), "the writer is named")
+	require.NotContains(t, err.Error(), holder.ID().String(), "the holder is not")
+}

internal/db/merge.go

@@ -15,6 +15,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"strings"
 	"sync"
 
 	"github.com/ipfs/go-cid"
@@ -42,6 +43,7 @@ import (
 func (db *DB) Merge(ctx context.Context, evt event.Merge) error {
 	col, err := getCollectionFromCollectionID(ctx, db, evt.CollectionID)
 	if err != nil {
+		db.stats.markDropped(collectionDropReason(err))
 		return err
 	}
 
@@ -61,21 +63,32 @@ func (db *DB) Merge(ctx context.Context, evt event.Merge) error {
 	for i := 0; i < db.txnAttempts(); i++ {
 		err = db.executeMerge(ctx, col, evt)
 		if errors.Is(err, corekv.ErrTxnConflict) {
+			db.stats.txnConflicts.Add(1)
 			continue
 		}
 		if err != nil {
+			db.stats.markDropped(mergeDropReason(err))
 			return err
 		}
 		return nil
 	}
-	return client.NewErrMaxTxnRetries(err)
+	// A single event has no smaller write set to retry over, so exhaustion is a drop rather than
+	// a fallback into per-event isolation.
+	db.stats.markDropped(dropRetryExhausted)
+	return NewErrMergeEventDropped(client.NewErrMaxTxnRetries(err), evt.DocID, evt.Cid.String())
 }
 
 // mergeChunkSize bounds how many events share a transaction. Badger re-sorts the
 // transaction's pending writes on every iterator open, and each merged document opens
 // 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 +119,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(collectionDropReason(err))
 			continue
 		}
 		entries = append(entries, mergeEntry{evt: evt, col: col, index: i})
@@ -178,6 +192,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 +212,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.txnConflicts.Add(1)
 				continue
 			}
 			return mergeErr
@@ -229,15 +265,30 @@ func (db *DB) mergeChunk(ctx context.Context, entries []mergeEntry) error {
 			txn.Discard()
 			if errors.Is(err, corekv.ErrTxnConflict) {
 				conflictErr = err
+				db.stats.txnConflicts.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.
+	// A chunk of one has no smaller write set for the caller to fall back to, so its exhaustion
+	// is a drop the caller records rather than a chunk to count here.
+	if len(entries) > 1 {
+		db.stats.chunkExhausted.Add(1)
+
+		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 +305,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 +322,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