perf: cache column UNIQUE ownership with commit-ordered publication (#20)

README.md

@@ -539,6 +539,14 @@ CREATE UNIQUE INDEX idx_users_email ON users(email);
 DROP INDEX idx_users_email;
 ```
 
+Column-based UNIQUE constraints build a derived ownership cache once from durable
+rows, then validate and update it under the table's exclusive publication gate.
+Changes are prepared before commit and published only after durable success.
+Rollback, compare failure, DDL, restart, and ambiguous transport errors never
+leave speculative ownership entries. NULL-bearing keys remain exempt; composite
+keys use the constraint encoder, not the SELECT candidate encoder. Expression
+UNIQUE constraints retain scan validation until evaluator determinism is proven.
+
 **PRAGMA:**
 ```sql
 PRAGMA table_info(users);

pkg/executor/unique_cache_test.go

@@ -0,0 +1,135 @@
+package executor
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+	"github.com/danfragoso/pizzasql-next/pkg/testkv"
+)
+
+func TestWarmUniqueValidationDoesNotScan(t *testing.T) {
+	kv := testkv.New(t)
+	pool := kv.Pool(8)
+	defer pool.Close()
+	schemas := storage.NewSchemaManager(pool, "unique_cache")
+	tables := storage.NewTableManager(pool, schemas, "unique_cache")
+	e := newExec(schemas, tables)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER UNIQUE)")
+	var values []string
+	for i := 1; i <= 100; i++ {
+		values = append(values, fmt.Sprintf("(%d,%d)", i, i))
+	}
+	execMust(t, e, "INSERT INTO t VALUES "+strings.Join(values, ","))
+	for _, sql := range []string{
+		"INSERT INTO t VALUES (101,101)",
+		"UPDATE t SET v=102 WHERE id=101",
+		"DELETE FROM t WHERE id=101",
+		"INSERT INTO t VALUES (102,102)",
+		"BEGIN", "INSERT INTO t VALUES (103,103)", "ROLLBACK",
+		"INSERT INTO t VALUES (103,103)",
+		"BEGIN", "UPDATE t SET v=501 WHERE id=1", "UPDATE t SET v=1 WHERE id=2", "UPDATE t SET v=2 WHERE id=1", "COMMIT",
+		"INSERT INTO t VALUES (104,NULL),(105,NULL)",
+	} {
+		before := kv.RequestCount(10)
+		execMust(t, e, sql)
+		if n := kv.RequestCount(10) - before; n != 0 {
+			t.Fatalf("%s: unique validation scanned %d pages", sql, n)
+		}
+	}
+	for _, sql := range []string{"INSERT INTO t VALUES (200,2)", "UPDATE t SET v=2 WHERE id=103"} {
+		if _, err := execSQL(e, sql); err == nil || !strings.Contains(err.Error(), "UNIQUE") {
+			t.Fatalf("%s: %v", sql, err)
+		}
+	}
+	execMust(t, e, "UPDATE t SET id=1000 WHERE id=1")
+	if _, err := execSQL(e, "INSERT INTO t VALUES (200,2)"); err == nil {
+		t.Fatal("PK move released an owned unique value")
+	}
+	execMust(t, e, "DELETE FROM t WHERE id=1000")
+	execMust(t, e, "INSERT INTO t VALUES (200,2)")
+
+	// A new manager models restart: durable rows rebuild the cache once, not on
+	// every mutation. Managers are shared per database in normal operation.
+	schemas = storage.NewSchemaManager(pool, "unique_cache")
+	tables = storage.NewTableManager(pool, schemas, "unique_cache")
+	e = newExec(schemas, tables)
+	before := kv.RequestCount(10)
+	execMust(t, e, "INSERT INTO t VALUES (201,201)")
+	if kv.RequestCount(10) == before {
+		t.Fatal("cold UNIQUE cache did not read durable state")
+	}
+	before = kv.RequestCount(10)
+	execMust(t, e, "INSERT INTO t VALUES (202,202)")
+	if kv.RequestCount(10) != before {
+		t.Fatal("warm UNIQUE cache rebuilt after restart")
+	}
+	if _, err := execSQL(e, "INSERT INTO t VALUES (203,201)"); err == nil {
+		t.Fatal("restarted cache lost an existing value")
+	}
+}
+
+func TestUniqueCacheDropRecreateAndFailedTransaction(t *testing.T) {
+	_, schemas, tables := newTestDB(t)
+	e := newExec(schemas, tables)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
+	execMust(t, e, "CREATE UNIQUE INDEX MixedCase ON t(v)")
+	execMust(t, e, "INSERT INTO t VALUES (1,1)")
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "INSERT INTO t VALUES (2,2),(3,2)")
+	if _, err := execSQL(e, "COMMIT"); err == nil {
+		t.Fatal("duplicate pending values committed")
+	}
+	execMust(t, e, "INSERT INTO t VALUES (2,2)")
+	execMust(t, e, "DROP INDEX MixedCase")
+	execMust(t, e, "INSERT INTO t VALUES (3,2)")
+	if _, err := execSQL(e, "CREATE UNIQUE INDEX MixedCase ON t(v)"); err == nil {
+		t.Fatal("recreated unique index ignored durable duplicate")
+	}
+	execMust(t, e, "DELETE FROM t WHERE id=3")
+	execMust(t, e, "CREATE UNIQUE INDEX MixedCase ON t(v)")
+	if _, err := execSQL(e, "INSERT INTO t VALUES (3,2)"); err == nil {
+		t.Fatal("recreated cache lost owner")
+	}
+}
+
+func TestUniqueCacheAmbiguousWriteOutcome(t *testing.T) {
+	for _, operation := range []string{"transaction", "insert", "update", "delete"} {
+		t.Run(operation, func(t *testing.T) {
+			kv := testkv.New(t)
+			pool := kv.Pool(8)
+			defer pool.Close()
+			schemas := storage.NewSchemaManager(pool, "ambiguous")
+			tables := storage.NewTableManager(pool, schemas, "ambiguous")
+			e := newExec(schemas, tables)
+			execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER UNIQUE)")
+			execMust(t, e, "INSERT INTO t VALUES (1,1)")
+			kv.DropNextCompareReply()
+			var err error
+			switch operation {
+			case "transaction":
+				_, err = execSQL(e, "INSERT INTO t VALUES (2,2)")
+			case "insert":
+				err = tables.Insert("t", storage.Row{"id": int64(2), "v": int64(2)})
+			case "update":
+				_, _, err = tables.UpdateByPK("t", "1", func(storage.Row) (storage.Row, error) { return storage.Row{"v": int64(2)}, nil })
+			case "delete":
+				_, _, err = tables.DeleteByPK("t", "1")
+			}
+			if err == nil {
+				t.Fatal("dropped reply unexpectedly succeeded")
+			}
+			if operation == "delete" {
+				execMust(t, e, "INSERT INTO t VALUES (3,1)")
+			} else {
+				if _, err := execSQL(e, "INSERT INTO t VALUES (3,2)"); err == nil {
+					t.Fatal("unknown but durable unique owner was ignored")
+				}
+				if operation == "update" {
+					execMust(t, e, "INSERT INTO t VALUES (4,1)")
+				}
+			}
+		})
+	}
+}

pkg/storage/table.go

@@ -32,6 +32,7 @@ type TableManager struct {
 	database string
 
 	cacheMu     sync.RWMutex
+	uniqueCache map[string]*uniqueIndexCache
 	indexCache  map[string]map[string][]int64 // index name → indexed value → rowids
 	indexTable  map[string]string             // index name → table name
 	rowKeyCache map[string]map[int64]string   // table name → rowid → primary key
@@ -90,6 +91,7 @@ func NewTableManager(pool *KVPool, schema *SchemaManager, database string) *Tabl
 		pool:                 pool,
 		schema:               schema,
 		database:             database,
+		uniqueCache:          make(map[string]*uniqueIndexCache),
 		indexCache:           make(map[string]map[string][]int64),
 		indexTable:           make(map[string]string),
 		rowKeyCache:          make(map[string]map[int64]string),
@@ -219,6 +221,13 @@ func (m *TableManager) invalidateCache(table string) {
 // InvalidateCache is the exported version for use by the executor.
 func (m *TableManager) InvalidateCache(table string) {
 	m.invalidateCache(table)
+	m.cacheMu.Lock()
+	for name, cache := range m.uniqueCache {
+		if cache.table == strings.ToLower(table) {
+			delete(m.uniqueCache, name)
+		}
+	}
+	m.cacheMu.Unlock()
 }
 
 // rowVisitFunc is invoked for each decoded row in a streaming scan. Return
@@ -575,8 +584,10 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 		return 0, err
 	}
 	defer unlock()
+	var uniquePlan uniqueCachePlan
 	if uniq {
-		if err := m.validateUniqueRows(table, []Row{nr}, nil); err != nil {
+		uniquePlan, err = m.prepareUniqueRows(table, map[string]*overlayEntry{key: {row: nr}})
+		if err != nil {
 			return 0, err
 		}
 	}
@@ -588,12 +599,14 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 		[]BatchOp{{Op: batchPut, Key: []byte(key), Value: data}},
 	)
 	if err != nil {
+		m.invalidateWriteFailure(table)
 		return 0, err
 	}
 	if !committed {
 		return 0, fmt.Errorf("duplicate primary key: %v", row[schemaPrimaryKey(m.schema, table)])
 	}
 
+	m.applyUniquePlan(uniquePlan, nil)
 	if err := m.updateIndexesForRow(table, nr, true); err != nil {
 		return 0, err
 	}
@@ -800,9 +813,21 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 		}
 	}
 
-	if err := m.validateUniqueRows(table, encoded, nil); err != nil {
+	var uniquePlan uniqueCachePlan
+	unique, err := m.hasUniqueIndex(table)
+	if err != nil {
 		return 0, 0, err
 	}
+	if unique {
+		uniqueEntries := make(map[string]*overlayEntry, len(encoded))
+		for i, row := range encoded {
+			uniqueEntries[string(ops[i].Key)] = &overlayEntry{row: row}
+		}
+		uniquePlan, err = m.prepareUniqueRows(table, uniqueEntries)
+		if err != nil {
+			return 0, 0, err
+		}
+	}
 
 	// Write rows in atomic BATCH_WRITE chunks. Maintain in-memory indexes only
 	// for rows that actually persisted, so a partial failure cannot leave an
@@ -823,6 +848,13 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 		numOK += len(chunk)
 	}
 	// Maintain indexes for the rows that persisted (the first numOK ops).
+	if len(uniquePlan) != 0 {
+		accepted := make(map[string]bool, numOK)
+		for i := 0; i < numOK; i++ {
+			accepted[string(ops[i].Key)] = true
+		}
+		m.applyUniquePlan(uniquePlan, accepted)
+	}
 	for i := 0; i < numOK; i++ {
 		if err := m.updateIndexesForRow(table, encoded[i], true); err != nil {
 			return numOK, 0, err
@@ -835,6 +867,9 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 		}
 	}
 	m.incrCount(table, schema.CreatedAt, numOK, wasInit)
+	if firstErr != nil {
+		m.invalidateWriteFailure(table)
+	}
 
 	var lastRowID int64
 	if numOK > 0 {
@@ -1057,8 +1092,10 @@ func (m *TableManager) UpdateByPK(table, pk string, updateFn func(Row) (Row, err
 	if err != nil {
 		return nil, false, err
 	}
+	var uniquePlan uniqueCachePlan
 	if uniq {
-		if err := m.validateUniqueRows(table, []Row{row}, excludedKey(key)); err != nil {
+		uniquePlan, err = m.prepareUniqueRows(table, map[string]*overlayEntry{key: {row: row}})
+		if err != nil {
 			return nil, false, err
 		}
 	}
@@ -1067,12 +1104,14 @@ func (m *TableManager) UpdateByPK(table, pk string, updateFn func(Row) (Row, err
 		[]BatchOp{{Op: batchPut, Key: []byte(key), Value: data}},
 	)
 	if err != nil {
+		m.invalidateWriteFailure(table)
 		return nil, false, err
 	}
 	if !committed {
 		return nil, false, ErrSerialization
 	}
 
+	m.applyUniquePlan(uniquePlan, nil)
 	if err := m.updateIndexesForRow(table, oldRow, false); err != nil {
 		return nil, false, err
 	}
@@ -1133,12 +1172,14 @@ func (m *TableManager) DeleteByPK(table, pk string) (Row, bool, error) {
 		[]BatchOp{{Op: batchDelete, Key: []byte(key)}},
 	)
 	if err != nil {
+		m.invalidateWriteFailure(table)
 		return nil, false, err
 	}
 	if !committed {
 		return nil, false, ErrSerialization
 	}
 
+	m.removeUniqueKey(table, key)
 	if err := m.updateIndexesForRow(table, row, false); err != nil {
 		return nil, false, err
 	}
@@ -1449,6 +1490,7 @@ func (m *TableManager) ClearIndex(indexName, tableName string, columns []string)
 	m.cacheMu.Lock()
 	delete(m.indexCache, indexKey)
 	delete(m.indexTable, indexKey)
+	delete(m.uniqueCache, indexKey)
 	m.disabledIndexes[indexKey] = true
 	rowKeysNeeded := false
 	for _, indexedTable := range m.indexTable {
@@ -1683,98 +1725,6 @@ func encodeIntegralFloat(v float64) (string, bool) {
 	return "i" + strconv.FormatInt(int64(v), 10), true
 }
 
-// validateUniqueRows checks that pending rows do not conflict with one another or
-// with durable rows on any UNIQUE index. excludedKeys lists the actual durable
-// data keys whose rows are being replaced in this operation (delete, or an update
-// of a non-indexed column), so those rows' own unique values do not self-conflict
-// and swapping two unique values remains possible. Matching is done against the
-// raw scanned KV key rather than a re-stringified primary key, so numeric and
-// composite keys are unambiguous. The caller must hold the table's exclusive gate.
-func (m *TableManager) validateUniqueRows(table string, pending []Row, excludedKeys map[string]bool) error {
-	indexes, err := m.uniqueIndexes(table)
-	if err != nil {
-		return err
-	}
-	if len(indexes) == 0 {
-		return nil
-	}
-
-	// seen tracks, per unique index, the encoded values already claimed by the
-	// pending rows. Keying by index name (not just the encoded value) is essential:
-	// a single row can legitimately carry the same value in two different unique
-	// indexes (e.g. name and lower_name both "alice"), which must not collide with
-	// itself. Two rows only conflict when they share a value within the SAME index.
-	seen := make(map[string]map[string]bool, len(indexes))
-	for _, row := range pending {
-		for _, idx := range indexes {
-			encoded, isNull, err := m.encodeUniqueValue(row, idx.Columns)
-			if err != nil {
-				return err
-			}
-			if isNull {
-				continue
-			}
-			perIndex := seen[idx.Name]
-			if perIndex == nil {
-				perIndex = make(map[string]bool)
-				seen[idx.Name] = perIndex
-			}
-			if perIndex[encoded] {
-				return &UniqueViolationError{Constraint: idx.Name}
-			}
-			perIndex[encoded] = true
-		}
-	}
-
-	return m.pool.WithClient(func(client *KVClient) (retErr error) {
-		cursor, err := client.ScanWithLimit([]byte(m.dataPrefix(table)), scanPageSize)
-		if err != nil {
-			return err
-		}
-		defer func() {
-			if err := cursor.Close(); retErr == nil {
-				retErr = err
-			}
-		}()
-		for {
-			entries, done, err := cursor.Next()
-			if err != nil {
-				return err
-			}
-			for _, e := range entries {
-				if excludedKeys != nil && excludedKeys[string(e.Key)] {
-					continue
-				}
-				row, err := decodeRow(e.Value)
-				if err != nil {
-					return err
-				}
-				for _, idx := range indexes {
-					encoded, isNull, encErr := m.encodeUniqueValue(row, idx.Columns)
-					if encErr != nil {
-						return encErr
-					}
-					if isNull {
-						continue
-					}
-					if perIndex := seen[idx.Name]; perIndex != nil && perIndex[encoded] {
-						return &UniqueViolationError{Constraint: idx.Name}
-					}
-				}
-			}
-			if done {
-				return nil
-			}
-		}
-	})
-}
-
-// excludedKey returns the exclusion set that exempts a single row's durable value
-// from the uniqueness scan, keyed by its actual durable data key.
-func excludedKey(dataKey string) map[string]bool {
-	return map[string]bool{dataKey: true}
-}
-
 // lockForWrite acquires the table gate appropriate for a point write: exclusive
 // when the table has a UNIQUE index (so the validating scan cannot race another
 // writer), shared otherwise. The uniqueness decision is re-checked under the lock

pkg/storage/tx.go

@@ -196,24 +196,14 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 	// durable rows. This runs under the exclusive gate acquired above, so it
 	// cannot race a concurrent writer. Duplicate unique values fail the commit
 	// with a "UNIQUE constraint failed" error rather than ErrSerialization.
+	uniquePlans := make(map[string]uniqueCachePlan, len(uniqueTables))
 	for t := range uniqueTables {
-		entries := s.overlay[t]
-		pending := make([]Row, 0, len(entries))
-		// Exclude every overlay data key: rows being written, updated, or deleted
-		// are all replaced by this transaction, so their durable unique values
-		// must not self-conflict with the staged state (e.g. delete a row and
-		// re-insert the same primary key with a different rowid).
-		excluded := make(map[string]bool, len(entries))
-		for key, e := range entries {
-			excluded[key] = true
-			if !e.absent {
-				pending = append(pending, e.row)
-			}
-		}
-		if err := s.table.validateUniqueRows(t, pending, excluded); err != nil {
+		plan, err := s.table.prepareUniqueRows(t, s.overlay[t])
+		if err != nil {
 			s.resetLocked()
 			return 0, err
 		}
+		uniquePlans[t] = plan
 	}
 
 	// Build the compare set from every observed key.
@@ -255,6 +245,13 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 		committed = ok
 		return err
 	})
+	if err != nil || !committed {
+		// A transport error can hide a durable commit; a compare conflict can
+		// also expose an out-of-band writer. Rebuild derived state next time.
+		for table := range s.overlay {
+			s.table.invalidateWriteFailure(table)
+		}
+	}
 	if err != nil {
 		s.resetLocked()
 		return 0, err
@@ -266,7 +263,8 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 
 	// Advance generations and invalidate derived caches for written tables.
 	for t := range s.overlay {
-		s.table.InvalidateCache(t)
+		s.table.invalidateCache(t)
+		s.table.applyUniquePlan(uniquePlans[t], nil)
 		s.table.bumpIndexPredicateWildcard(t)
 		s.table.bumpGeneration(t)
 	}

pkg/storage/unique_cache.go

@@ -0,0 +1,207 @@
+package storage
+
+import "strings"
+
+// UNIQUE caches use the constraint encoder, not the coarser SELECT candidate
+// encoder. Owners are actual durable keys, preserving NULL and legacy-key rules.
+// Validation and publication hold the table's exclusive gate. cacheMu protects
+// cache membership and removals from shared-gate point deletes.
+type uniqueIndexCache struct {
+	table  string
+	values map[string]map[string]bool
+	keys   map[string]string
+}
+
+type uniqueCacheChange struct {
+	name   string
+	cache  *uniqueIndexCache
+	values map[string]*string // nil removes a key or represents a NULL index value
+}
+
+type uniqueCachePlan []uniqueCacheChange
+
+func (m *TableManager) prepareUniqueRows(table string, entries map[string]*overlayEntry) (uniqueCachePlan, error) {
+	indexes, err := m.uniqueIndexes(table)
+	if err != nil {
+		return nil, err
+	}
+	if len(indexes) == 0 {
+		return nil, nil
+	}
+	caches, err := m.ensureUniqueCaches(table, indexes)
+	if err != nil {
+		return nil, err
+	}
+	plan := make(uniqueCachePlan, 0, len(indexes))
+	for _, idx := range indexes {
+		cache := caches[idx.Name]
+		change := uniqueCacheChange{name: strings.ToLower(idx.Name), cache: cache, values: make(map[string]*string, len(entries))}
+		seen := make(map[string]bool)
+		for key, entry := range entries {
+			change.values[key] = nil
+			if entry.absent {
+				continue
+			}
+			value, isNull, err := m.encodeUniqueValue(entry.row, idx.Columns)
+			if err != nil {
+				return nil, err
+			}
+			if isNull {
+				continue
+			}
+			if seen[value] {
+				return nil, &UniqueViolationError{Constraint: idx.Name}
+			}
+			seen[value] = true
+			for owner := range cache.values[value] {
+				if _, replaced := entries[owner]; !replaced {
+					return nil, &UniqueViolationError{Constraint: idx.Name}
+				}
+			}
+			change.values[key] = &value
+		}
+		plan = append(plan, change)
+	}
+	return plan, nil
+}
+
+func (m *TableManager) ensureUniqueCaches(table string, indexes []*Index) (map[string]*uniqueIndexCache, error) {
+	caches := make(map[string]*uniqueIndexCache, len(indexes))
+	var missing []*Index
+	m.cacheMu.RLock()
+	for _, idx := range indexes {
+		if cache := m.uniqueCache[strings.ToLower(idx.Name)]; cache != nil && cacheableUniqueIndex(idx) {
+			caches[idx.Name] = cache
+		} else {
+			missing = append(missing, idx)
+		}
+	}
+	m.cacheMu.RUnlock()
+	if len(missing) == 0 {
+		return caches, nil
+	}
+	for _, idx := range missing {
+		caches[idx.Name] = &uniqueIndexCache{table: strings.ToLower(table), values: make(map[string]map[string]bool), keys: make(map[string]string)}
+	}
+	err := m.pool.WithClient(func(client *KVClient) (retErr error) {
+		cursor, err := client.ScanWithLimit([]byte(m.dataPrefix(table)), scanPageSize)
+		if err != nil {
+			return err
+		}
+		defer func() {
+			if err := cursor.Close(); retErr == nil {
+				retErr = err
+			}
+		}()
+		for {
+			entries, done, err := cursor.Next()
+			if err != nil {
+				return err
+			}
+			for _, entry := range entries {
+				row, err := decodeRow(entry.Value)
+				if err != nil {
+					return err
+				}
+				for _, idx := range missing {
+					value, isNull, err := m.encodeUniqueValue(row, idx.Columns)
+					if err != nil {
+						return err
+					}
+					if !isNull {
+						caches[idx.Name].add(string(entry.Key), value)
+					}
+				}
+			}
+			if done {
+				return nil
+			}
+		}
+	})
+	if err != nil {
+		return nil, err
+	}
+	m.cacheMu.Lock()
+	for _, idx := range missing {
+		if cacheableUniqueIndex(idx) {
+			m.uniqueCache[strings.ToLower(idx.Name)] = caches[idx.Name]
+		}
+	}
+	m.cacheMu.Unlock()
+	return caches, nil
+}
+
+// Expression evaluators have no determinism contract. Rebuild these constraints
+// for each validation rather than freezing time/random/function-dependent values.
+func cacheableUniqueIndex(index *Index) bool {
+	for _, column := range index.Columns {
+		if column.Expression != "" {
+			return false
+		}
+	}
+	return true
+}
+
+func (c *uniqueIndexCache) remove(key string) {
+	value, ok := c.keys[key]
+	if !ok {
+		return
+	}
+	delete(c.keys, key)
+	delete(c.values[value], key)
+	if len(c.values[value]) == 0 {
+		delete(c.values, value)
+	}
+}
+
+func (c *uniqueIndexCache) add(key, value string) {
+	if c.values[value] == nil {
+		c.values[value] = make(map[string]bool)
+	}
+	c.values[value][key] = true
+	c.keys[key] = value
+}
+
+// Apply only after durable success. accepted is used by the storage bulk API,
+// which can commit several batches; SQL statements use one atomic transaction.
+func (m *TableManager) applyUniquePlan(plan uniqueCachePlan, accepted map[string]bool) {
+	if len(plan) == 0 {
+		return
+	}
+	m.cacheMu.Lock()
+	defer m.cacheMu.Unlock()
+	for _, change := range plan {
+		if m.uniqueCache[change.name] != change.cache {
+			continue
+		}
+		for key := range change.values {
+			if accepted == nil || accepted[key] {
+				change.cache.remove(key)
+			}
+		}
+		for key, value := range change.values {
+			if value != nil && (accepted == nil || accepted[key]) {
+				change.cache.add(key, *value)
+			}
+		}
+	}
+}
+
+// A transport failure may hide a durable write. Drop derived state and make
+// earlier predicate/scan snapshots retry rather than trusting an unknown outcome.
+// Callers retain their normal table publication gate.
+func (m *TableManager) invalidateWriteFailure(table string) {
+	m.InvalidateCache(table)
+	m.bumpIndexPredicateWildcard(table)
+	m.bumpGeneration(table)
+}
+
+func (m *TableManager) removeUniqueKey(table, key string) {
+	m.cacheMu.Lock()
+	defer m.cacheMu.Unlock()
+	for _, cache := range m.uniqueCache {
+		if cache.table == strings.ToLower(table) {
+			cache.remove(key)
+		}
+	}
+}

pkg/storage/unique_cache_test.go

@@ -0,0 +1,65 @@
+package storage
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+)
+
+func TestUniqueCacheNotPublishedOnCompareConflict(t *testing.T) {
+	_, _, schemas, tables := newTestSession(t)
+	createTestTable(t, schemas, "t", []Column{{Name: "id", Type: "INTEGER", PrimaryKey: true}, {Name: "v", Type: "INTEGER"}})
+	if err := tables.CreateUniqueIndex(&Index{Name: "unique_v", Table: "t", Unique: true, Columns: []IndexColumn{{Name: "v"}}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := tables.Insert("t", Row{"id": int64(1), "v": int64(1)}); err != nil {
+		t.Fatal(err)
+	}
+	session := NewSession(schemas, tables)
+	if err := session.Begin(); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := session.GetByPK("t", "1"); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := session.UpdateByPK("t", "1", func(Row) (Row, error) { return Row{"v": int64(2)}, nil }); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := tables.UpdateByPK("t", "1", func(Row) (Row, error) { return Row{"v": int64(3)}, nil }); err != nil {
+		t.Fatal(err)
+	}
+	if err := session.Commit(); !errors.Is(err, ErrSerialization) {
+		t.Fatalf("commit: %v", err)
+	}
+	if err := tables.Insert("t", Row{"id": int64(2), "v": int64(2)}); err != nil {
+		t.Fatalf("failed commit claimed a value: %v", err)
+	}
+	if err := tables.Insert("t", Row{"id": int64(3), "v": int64(3)}); err == nil {
+		t.Fatal("failed commit lost durable owner")
+	}
+}
+
+func TestUniqueExpressionCacheDoesNotFreezeEvaluator(t *testing.T) {
+	_, _, schemas, tables := newTestSession(t)
+	createTestTable(t, schemas, "t", []Column{{Name: "id", Type: "INTEGER", PrimaryKey: true}, {Name: "v", Type: "TEXT"}})
+	current := false
+	tables.SetExpressionEvaluator(func(_ string, row Row) (interface{}, error) {
+		if current {
+			return "same", nil
+		}
+		return fmt.Sprint(row["v"]), nil
+	})
+	if err := tables.CreateUniqueIndex(&Index{Name: "expr", Table: "t", Unique: true, Columns: []IndexColumn{{Expression: "value(v)"}}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := tables.Insert("t", Row{"id": int64(1), "v": "a"}); err != nil {
+		t.Fatal(err)
+	}
+	current = true
+	if err := tables.Insert("t", Row{"id": int64(2), "v": "b"}); err == nil {
+		t.Fatal("constraint froze expression evaluation")
+	}
+	if len(tables.uniqueCache) != 0 {
+		t.Fatal("expression UNIQUE cache retained without determinism proof")
+	}
+}

pkg/testkv/testkv.go

@@ -53,11 +53,12 @@ type scan struct {
 
 // Server is an in-memory KV server.
 type Server struct {
-	mu       sync.Mutex
-	data     map[string][]byte
-	lsns     map[string]uint64
-	nextLSN  uint64
-	requests map[uint16]int
+	mu               sync.Mutex
+	data             map[string][]byte
+	lsns             map[string]uint64
+	nextLSN          uint64
+	requests         map[uint16]int
+	dropCompareReply bool
 
 	ln       net.Listener
 	wg       sync.WaitGroup
@@ -134,12 +135,29 @@ func (s *Server) handle(conn net.Conn) {
 			return
 		}
 		body := s.execute(opcode, payload)
+		s.mu.Lock()
+		drop := opcode == opCompareBatch && s.dropCompareReply
+		if drop {
+			s.dropCompareReply = false
+		}
+		s.mu.Unlock()
+		if drop {
+			return
+		}
 		if _, err := conn.Write(encodeResponse(opcode, requestID, body)); err != nil {
 			return
 		}
 	}
 }
 
+// DropNextCompareReply commits the next compare request but closes its connection
+// before replying, simulating an ambiguous durable write outcome.
+func (s *Server) DropNextCompareReply() {
+	s.mu.Lock()
+	s.dropCompareReply = true
+	s.mu.Unlock()
+}
+
 // RequestCount reports observed PKBFI requests for bounded-work assertions.
 func (s *Server) RequestCount(opcode uint16) int {
 	s.mu.Lock()