chore: conflict and ingest instrumentation

Author: t

internal/datastore/readkinds.go

@@ -0,0 +1,145 @@
+// 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 datastore
+
+import (
+	"strings"
+	"sync/atomic"
+)
+
+// ReadKind names one class of key a transaction's read set can hold.
+//
+// The underlying store reports a commit conflict without naming the contended key, so the
+// set of classes a transaction read is the only lead available for working out what it
+// contended with. The classes are bit flags so a whole transaction's set fits in one word
+// and recording a read costs an atomic OR with no allocation.
+type ReadKind uint32
+
+const (
+	// ReadDoc is a document field value or its priority marker, under the datastore.
+	ReadDoc ReadKind = 1 << iota
+	// ReadIndex is a secondary index entry, under the datastore. Set by the index write
+	// path, which is the only place that knows an encoded datastore key is an index entry.
+	ReadIndex
+	// ReadUniqueIndex is the existence check a unique index performs before writing its
+	// entry. Set by the unique index write path for the same reason as ReadIndex.
+	ReadUniqueIndex
+	// ReadHead is a headstore key: a document's or collection's current DAG heads.
+	ReadHead
+	// ReadBlock is a blockstore key: an IPLD block or its to-merge marker.
+	ReadBlock
+	// ReadSystem is a systemstore key that is not a sequence.
+	ReadSystem
+	// ReadSequence is a systemstore /seq/ key. Short ID allocation commits the sequence in
+	// its own transaction, so a merge never carries one in its read set and this stays zero
+	// on that path. It is kept for callers that do read a sequence inline.
+	ReadSequence
+	// ReadPeer is a peerstore key.
+	ReadPeer
+	// ReadEnc is an encryption keystore key.
+	ReadEnc
+	// ReadRangeIterator means the transaction opened an iterator bounded by start and end
+	// rather than by prefix. Badger's prefix option is unset for those, so its bounds
+	// check reads the first key past the range into the read set.
+	ReadRangeIterator
+	// ReadPrefixIterator means the transaction opened a prefix-bounded iterator, which
+	// cannot read past its range.
+	ReadPrefixIterator
+)
+
+// readKindNames is ordered to match the bit order above.
+var readKindNames = [...]string{
+	"doc",
+	"index",
+	"uniqueIndex",
+	"head",
+	"block",
+	"system",
+	"sequence",
+	"peer",
+	"enc",
+	"rangeIter",
+	"prefixIter",
+}
+
+// ReadKindCount is the number of distinct kinds, for callers keeping a counter per kind.
+const ReadKindCount = len(readKindNames)
+
+// ReadKindName returns the name of the kind held in bit i, or "" if i is out of range.
+func ReadKindName(i int) string {
+	if i < 0 || i >= len(readKindNames) {
+		return ""
+	}
+	return readKindNames[i]
+}
+
+// String renders the set as a pipe-separated list in bit order, e.g. "doc|head|rangeIter".
+// An empty set renders as "none".
+func (k ReadKind) String() string {
+	if k == 0 {
+		return "none"
+	}
+	var b strings.Builder
+	for i, name := range readKindNames {
+		if k&(1<<uint(i)) == 0 {
+			continue
+		}
+		if b.Len() > 0 {
+			b.WriteByte('|')
+		}
+		b.WriteString(name)
+	}
+	return b.String()
+}
+
+// ReadKinds accumulates the kinds of key a single transaction has read.
+//
+// The zero value is ready to use. It is safe for concurrent use, which matters because a
+// transaction may be read from more than one goroutine even though a merge is not.
+type ReadKinds struct {
+	bits atomic.Uint32
+}
+
+// Mark records that a key of the given kind entered the read set. Safe on a nil receiver
+// so call sites do not have to check whether the transaction records kinds.
+func (r *ReadKinds) Mark(kind ReadKind) {
+	if r == nil {
+		return
+	}
+	r.bits.Or(uint32(kind))
+}
+
+// Kinds returns the set recorded so far.
+func (r *ReadKinds) Kinds() ReadKind {
+	if r == nil {
+		return 0
+	}
+	return ReadKind(r.bits.Load())
+}
+
+// readKindsCarrier is implemented by transactions that record read kinds. It is optional:
+// shims and mocks that do not record them are simply not asked.
+type readKindsCarrier interface {
+	ReadKinds() *ReadKinds
+}
+
+// ReadKindsOf returns the recorder for txn, or nil if txn does not record read kinds.
+// A nil result is usable: every method on ReadKinds tolerates it.
+//
+// Used by the few call sites that know a key's class from its Go type rather than its
+// encoded bytes: an index entry and a document value are both datastore keys and are
+// indistinguishable once encoded.
+func ReadKindsOf(txn any) *ReadKinds {
+	if c, ok := txn.(readKindsCarrier); ok {
+		return c.ReadKinds()
+	}
+	return nil
+}

internal/datastore/readkinds_test.go

@@ -0,0 +1,99 @@
+// 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 datastore
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/sourcenetwork/corekv"
+	"github.com/sourcenetwork/corekv/memory"
+	"github.com/sourcenetwork/immutable"
+
+	"github.com/sourcenetwork/defradb/internal/db/lock"
+	"github.com/sourcenetwork/defradb/internal/keys"
+)
+
+func TestKindForKey(t *testing.T) {
+	tests := []struct {
+		name string
+		key  string
+		want ReadKind
+	}{
+		{"datastore", "d/1/v/2/3", ReadDoc},
+		{"headstore", "h/1/2", ReadHead},
+		{"blockstore", "bsomecid", ReadBlock},
+		{"systemstore", "s/collection/name/Foo", ReadSystem},
+		{"sequence", "s/seq/doc", ReadSequence},
+		{"peerstore", "p/replicator/1", ReadPeer},
+		{"encstore", "esomecid", ReadEnc},
+		{"unknown", "?/1", 0},
+		{"empty", "", 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, kindForKey([]byte(tt.key)))
+		})
+	}
+}
+
+func TestReadKindString(t *testing.T) {
+	require.Equal(t, "none", ReadKind(0).String())
+	require.Equal(t, "doc", ReadDoc.String())
+	require.Equal(t, "doc|head|rangeIter", (ReadDoc | ReadHead | ReadRangeIterator).String())
+}
+
+// The recorder has to see the store discriminator that namespace.Wrap prepends, so these
+// go through a real transaction rather than calling the recorder directly.
+func TestTxnRecordsReadKinds(t *testing.T) {
+	ctx := context.Background()
+	txn := NewTxnFrom(memory.NewDatastore(ctx), lock.NewLockSet(), 0, false, immutable.None[int]())
+	defer txn.Discard()
+	ctx = CtxSetTxn(ctx, txn)
+
+	require.Equal(t, ReadKind(0), txn.ReadKinds().Kinds())
+
+	key := keys.DataStoreKey{CollectionShortID: 1, InstanceType: keys.ValueKey, DocShortID: 2, FieldID: "3"}
+	_, err := txn.Datastore().Get(ctx, key)
+	require.ErrorIs(t, err, corekv.ErrNotFound)
+	require.Equal(t, ReadDoc, txn.ReadKinds().Kinds())
+
+	_, err = txn.Headstore().Get(ctx, []byte("/1/2"))
+	require.ErrorIs(t, err, corekv.ErrNotFound)
+	require.Equal(t, ReadDoc|ReadHead, txn.ReadKinds().Kinds())
+}
+
+// A start/end iterator is the shape whose bounds check reads one key past the requested
+// range. Distinguishing it from a prefix iterator is the point of the two iterator bits.
+func TestTxnRecordsIteratorShape(t *testing.T) {
+	ctx := context.Background()
+
+	prefix := keys.DataStoreKey{CollectionShortID: 1, InstanceType: keys.ValueKey, DocShortID: 2}
+
+	rangeTxn := NewTxnFrom(memory.NewDatastore(ctx), lock.NewLockSet(), 0, false, immutable.None[int]())
+	defer rangeTxn.Discard()
+	iter, err := rangeTxn.Datastore().Iterator(
+		CtxSetTxn(ctx, rangeTxn),
+		IterOptions{Start: prefix, End: prefix.PrefixEnd()},
+	)
+	require.NoError(t, err)
+	require.NoError(t, iter.Close())
+	require.Equal(t, ReadDoc|ReadRangeIterator, rangeTxn.ReadKinds().Kinds())
+
+	prefixTxn := NewTxnFrom(memory.NewDatastore(ctx), lock.NewLockSet(), 0, false, immutable.None[int]())
+	defer prefixTxn.Discard()
+	iter, err = prefixTxn.Datastore().Iterator(CtxSetTxn(ctx, prefixTxn), IterOptions{Prefix: prefix})
+	require.NoError(t, err)
+	require.NoError(t, iter.Close())
+	require.Equal(t, ReadDoc|ReadPrefixIterator, prefixTxn.ReadKinds().Kinds())
+}

internal/datastore/readrecorder.go

@@ -0,0 +1,99 @@
+// 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 datastore
+
+import (
+	"bytes"
+	"context"
+
+	"github.com/sourcenetwork/corekv"
+)
+
+// sequencePrefix is the systemstore prefix under which all sequences live.
+var sequencePrefix = []byte("/seq/")
+
+// readRecorder sits between a transaction's namespaced stores and the raw transaction,
+// classifying every key the transaction reads by the store it belongs to.
+//
+// Writes pass through unclassified: the underlying store's conflict detection compares a
+// transaction's reads against other transactions' writes, so only reads decide whether a
+// commit conflicts.
+//
+// The recorder must not be handed to corekv as a context transaction. corekv casts the
+// context value to its own concrete transaction type, and a wrapper fails that cast.
+type readRecorder struct {
+	store corekv.ReaderWriter
+	kinds *ReadKinds
+}
+
+var _ corekv.ReaderWriter = (*readRecorder)(nil)
+
+// kindForKey classifies a rootstore key by its leading store discriminator byte, which
+// namespace.Wrap has already prepended by the time the key reaches here.
+func kindForKey(key []byte) ReadKind {
+	if len(key) == 0 {
+		return 0
+	}
+	switch key[0] {
+	case dataStoreKey:
+		// Index entries are also datastore keys and encode identically to document keys.
+		// The index write path marks those itself; everything else here is a document.
+		return ReadDoc
+	case headStoreKey:
+		return ReadHead
+	case blockStoreKey:
+		return ReadBlock
+	case systemStoreKey:
+		if bytes.HasPrefix(key[1:], sequencePrefix) {
+			return ReadSequence
+		}
+		return ReadSystem
+	case peerStoreKey:
+		return ReadPeer
+	case encStoreKey:
+		return ReadEnc
+	default:
+		return 0
+	}
+}
+
+func (r *readRecorder) Get(ctx context.Context, key []byte) ([]byte, error) {
+	r.kinds.Mark(kindForKey(key))
+	return r.store.Get(ctx, key)
+}
+
+func (r *readRecorder) Has(ctx context.Context, key []byte) (bool, error) {
+	r.kinds.Mark(kindForKey(key))
+	return r.store.Has(ctx, key)
+}
+
+// Iterator records the shape of the iterator as well as what it scans. A start/end
+// iterator is the shape whose bounds check reads one key past the requested range, so
+// separating the two shapes is what makes that over-read visible.
+func (r *readRecorder) Iterator(ctx context.Context, opts corekv.IterOptions) (corekv.Iterator, error) {
+	switch {
+	case opts.Prefix != nil:
+		r.kinds.Mark(ReadPrefixIterator | kindForKey(opts.Prefix))
+	case opts.Start != nil:
+		r.kinds.Mark(ReadRangeIterator | kindForKey(opts.Start))
+	case opts.End != nil:
+		r.kinds.Mark(ReadRangeIterator | kindForKey(opts.End))
+	}
+	return r.store.Iterator(ctx, opts)
+}
+
+func (r *readRecorder) Set(ctx context.Context, key, value []byte) error {
+	return r.store.Set(ctx, key, value)
+}
+
+func (r *readRecorder) Delete(ctx context.Context, key []byte) error {
+	return r.store.Delete(ctx, key)
+}

internal/datastore/txn.go

@@ -90,6 +90,10 @@ type BasicTxn struct {
 	id            uint64
 	ts            time.Time // timestamp
 
+	// readKinds records which classes of key this transaction's read set holds, so a
+	// commit conflict can be attributed to a class of key rather than reported bare.
+	readKinds ReadKinds
+
 	successFns []func()
 	errorFns   []func()
 	discardFns []func()
@@ -110,14 +114,26 @@ func NewTxnFrom(
 	chunkSize immutable.Option[int],
 ) *BasicTxn {
 	rootTxn := rootstore.NewTxn(readonly)
-	multistore := NewMultistore(rootTxn, lockSet, chunkSize)
 
-	return &BasicTxn{
-		Multistore:    multistore,
+	txn := &BasicTxn{
 		underlyingTxn: rootTxn,
 		id:            id,
 		ts:            time.Now(),
 	}
+	// The child stores read through the recorder; underlyingTxn stays unwrapped because
+	// corekv casts it to its own concrete type when it is fetched from a context.
+	txn.Multistore = NewMultistore(
+		&readRecorder{store: rootTxn, kinds: &txn.readKinds},
+		lockSet,
+		chunkSize,
+	)
+
+	return txn
+}
+
+// ReadKinds returns the classes of key this transaction has read so far.
+func (t *BasicTxn) ReadKinds() *ReadKinds {
+	return &t.readKinds
 }
 
 // The raw underlying txn.