Artifact [c7497b3b48]

Artifact c7497b3b48a68e9ae38fec8791c4e3cbc0e3cc4ab40be17a3661c5281defa844:

• File dst/tablesync_test.go — part of check-in [f5744359d3] at 2026-04-19 16:30:54 on branch trunk — Initial open source release

  libfossil is a pure-Go implementation of the Fossil SCM, reading and writing the same .fossil SQLite repository format as upstream Fossil.

  This is the first public release of code that previously lived in a private repository. The module path is github.com/danmestas/libfossil and the binary is named libfossil to avoid $PATH collisions with upstream fossil.

  Included: - Repo lifecycle (create, clone), checkout/checkin, timeline, merge, diff, annotate, shun, tag, branch, bisect, stash, undo/redo - Sync-over-HTTP protocol (client and handler) - Two SQLite drivers: modernc (pure Go, default) and ncruces (wasm-capable) - Observer interfaces (SyncObserver, CheckoutObserver) with an optional OpenTelemetry implementation in observer/otel (separate submodule) - Deterministic simulation testing harness with BUGGIFY fault injection - Kong-based embeddable CLI under cli/ and a standalone binary under cmd/libfossil/ (user: [email protected], size: 38989) [annotate] [blame] [check-ins using]

package dst

import (
	"fmt"
	"testing"
	"time"

	"github.com/danmestas/libfossil/internal/repo"
	"github.com/danmestas/libfossil/internal/uv"
)

// --- Table sync test helpers ---

func deviceTableDef() repo.TableDef {
	return repo.TableDef{
		Columns: []repo.ColumnDef{
			{Name: "device_id", Type: "text", PK: true},
			{Name: "hostname", Type: "text"},
			{Name: "status", Type: "text"},
		},
		Conflict: "mtime-wins",
	}
}

func configTableDef() repo.TableDef {
	return repo.TableDef{
		Columns: []repo.ColumnDef{
			{Name: "key", Type: "text", PK: true},
			{Name: "value", Type: "text"},
		},
		Conflict: "mtime-wins",
	}
}

func selfWriteTableDef() repo.TableDef {
	return repo.TableDef{
		Columns: []repo.ColumnDef{
			{Name: "peer_id", Type: "text", PK: true},
			{Name: "last_seen", Type: "integer"},
			{Name: "health", Type: "text"},
		},
		Conflict: "self-write",
	}
}

func ownerWriteTableDef() repo.TableDef {
	return repo.TableDef{
		Columns: []repo.ColumnDef{
			{Name: "resource_id", Type: "text", PK: true},
			{Name: "data", Type: "text"},
		},
		Conflict: "owner-write",
	}
}

func registerTable(t *testing.T, r *repo.Repo, name string, def repo.TableDef, mtime int64) {
	t.Helper()
	if err := repo.EnsureSyncSchema(r.DB()); err != nil {
		t.Fatalf("EnsureSyncSchema: %v", err)
	}
	if err := repo.RegisterSyncedTable(r.DB(), name, def, mtime); err != nil {
		t.Fatalf("RegisterSyncedTable %s: %v", name, err)
	}
}

func registerTableAll(t *testing.T, sim *Simulator, masterRepo *repo.Repo, name string, def repo.TableDef, mtime int64) {
	t.Helper()
	registerTable(t, masterRepo, name, def, mtime)
	for _, leafID := range sim.LeafIDs() {
		registerTable(t, sim.Leaf(leafID).Repo(), name, def, mtime)
	}
}

func upsertRow(t *testing.T, r *repo.Repo, table string, row map[string]any, mtime int64) {
	t.Helper()
	if err := repo.UpsertXRow(r.DB(), table, row, mtime); err != nil {
		t.Fatalf("UpsertXRow: %v", err)
	}
}

func assertRowCount(t *testing.T, r *repo.Repo, label, table string, def repo.TableDef, n int) {
	t.Helper()
	rows, _, err := repo.ListXRows(r.DB(), table, def)
	if err != nil {
		t.Fatalf("ListXRows %s: %v", label, err)
	}
	if len(rows) != n {
		t.Errorf("%s: expected %d rows, got %d", label, n, len(rows))
	}
}

func assertRowValue(t *testing.T, r *repo.Repo, label, table string, def repo.TableDef, pk map[string]any, col, want string) {
	t.Helper()
	var pkColDefs []repo.ColumnDef
	for _, c := range def.Columns {
		if c.PK {
			pkColDefs = append(pkColDefs, c)
		}
	}
	pkHash := repo.PKHash(pkColDefs, pk)
	row, _, err := repo.LookupXRow(r.DB(), table, def, pkHash)
	if err != nil {
		t.Fatalf("LookupXRow %s: %v", label, err)
	}
	if row == nil {
		t.Fatalf("%s: missing row pk=%v", label, pk)
	}
	got, _ := row[col].(string)
	if got != want {
		t.Errorf("%s: %s=%q, want %q", label, col, got, want)
	}
}

func logRowCounts(t *testing.T, sim *Simulator, table string, def repo.TableDef) {
	t.Helper()
	for _, leafID := range sim.LeafIDs() {
		rows, _, _ := repo.ListXRows(sim.Leaf(leafID).Repo().DB(), table, def)
		t.Logf("  %s: %d rows", leafID, len(rows))
	}
}

func newTableSyncSim(t *testing.T, seed int64, numLeaves int, buggify bool) (*Simulator, *repo.Repo, *MockFossil) {
	t.Helper()
	masterRepo := createMasterRepo(t)
	mf := NewMockFossil(masterRepo)
	sim, err := New(SimConfig{
		Seed:                seed,
		NumLeaves:           numLeaves,
		PollInterval:        5 * time.Second,
		TmpDir:              t.TempDir(),
		Upstream:            mf,
		Buggify:             buggify,
		SafetyCheckInterval: 10,
	})
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	t.Cleanup(func() { sim.Close() })
	sim.SetMasterRepo(masterRepo)
	return sim, masterRepo, mf
}

func runAndCheck(t *testing.T, sim *Simulator, sev severity, seed int64, steps int) {
	t.Helper()
	if err := sim.Run(steps); err != nil {
		t.Fatalf("Run: %v", err)
	}
	t.Logf("[%s] seed=%d steps=%d syncs=%d errors=%d",
		sev.Name, seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)
	if err := sim.CheckSafety(); err != nil {
		t.Fatalf("Safety: %v", err)
	}
}

func isNormalMode(sev severity) bool {
	return sev.DropRate == 0 && !sev.Buggify
}

// =============================================================================
// Level 1: Normal (0% faults)
// =============================================================================

func TestTS_Convergence3Peer(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(200)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "devices"
	def := deviceTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i, leafID := range sim.LeafIDs() {
		r := sim.Leaf(leafID).Repo()
		for j := 0; j < 3; j++ {
			upsertRow(t, r, table, map[string]any{
				"device_id": fmt.Sprintf("leaf%d-dev%d", i, j),
				"hostname":  fmt.Sprintf("host-%d-%d", i, j),
				"status":    "online",
			}, int64(100+i*10+j))
		}
	}

	runAndCheck(t, sim, sev, seed, stepsFor(300))

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
		for _, leafID := range sim.LeafIDs() {
			assertRowCount(t, sim.Leaf(leafID).Repo(), string(leafID), table, def, 9)
		}
		assertRowCount(t, masterRepo, "master", table, def, 9)
	} else {
		logRowCounts(t, sim, table, def)
	}
}

func TestTS_SchemaDeployChain(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(201)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "settings"
	def := configTableDef()
	registerTable(t, masterRepo, table, def, 100)

	runAndCheck(t, sim, sev, seed, stepsFor(300))

	if isNormalMode(sev) {
		for _, leafID := range sim.LeafIDs() {
			r := sim.Leaf(leafID).Repo()
			tables, err := repo.ListSyncedTables(r.DB())
			if err != nil {
				t.Fatalf("ListSyncedTables %s: %v", leafID, err)
			}
			found := false
			for _, tbl := range tables {
				if tbl.Name == table {
					found = true
				}
			}
			if !found {
				t.Errorf("%s: schema %q not deployed", leafID, table)
			}
		}
	}
}

func TestTS_MultipleTables5(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(202)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 2, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	defs := make([]repo.TableDef, 5)
	names := make([]string, 5)
	for i := 0; i < 5; i++ {
		names[i] = fmt.Sprintf("table_%d", i)
		defs[i] = configTableDef()
		registerTableAll(t, sim, masterRepo, names[i], defs[i], 100)
		for j := 0; j < 5; j++ {
			upsertRow(t, masterRepo, names[i], map[string]any{
				"key":   fmt.Sprintf("k%d-%d", i, j),
				"value": fmt.Sprintf("v%d-%d", i, j),
			}, 100)
		}
	}

	runAndCheck(t, sim, sev, seed, stepsFor(300))

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
		for i, name := range names {
			for _, leafID := range sim.LeafIDs() {
				assertRowCount(t, sim.Leaf(leafID).Repo(), string(leafID), name, defs[i], 5)
			}
		}
	}
}

func TestTS_CatalogHashShortCircuit(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(203)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 2, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "configs"
	def := configTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	allRepos := []*repo.Repo{masterRepo}
	for _, leafID := range sim.LeafIDs() {
		allRepos = append(allRepos, sim.Leaf(leafID).Repo())
	}
	for _, r := range allRepos {
		for i := 0; i < 10; i++ {
			upsertRow(t, r, table, map[string]any{
				"key":   fmt.Sprintf("k%d", i),
				"value": fmt.Sprintf("v%d", i),
			}, 100)
		}
	}

	runAndCheck(t, sim, sev, seed, stepsFor(200))

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
		masterHash, _ := repo.CatalogHash(masterRepo.DB(), table, def)
		for _, leafID := range sim.LeafIDs() {
			leafHash, _ := repo.CatalogHash(sim.Leaf(leafID).Repo().DB(), table, def)
			if leafHash != masterHash {
				t.Errorf("%s hash %s != master %s", leafID, leafHash, masterHash)
			}
		}
	}
}

func TestTS_MtimeWinsSamePK(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(204)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "config"
	def := configTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i, leafID := range sim.LeafIDs() {
		upsertRow(t, sim.Leaf(leafID).Repo(), table, map[string]any{
			"key":   "theme",
			"value": fmt.Sprintf("val-%d", i),
		}, int64(100+i*100))
	}

	runAndCheck(t, sim, sev, seed, stepsFor(300))

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
		pk := map[string]any{"key": "theme"}
		assertRowValue(t, masterRepo, "master", table, def, pk, "value", "val-2")
		for _, leafID := range sim.LeafIDs() {
			assertRowValue(t, sim.Leaf(leafID).Repo(), string(leafID), table, def, pk, "value", "val-2")
		}
	}
}

func TestTS_MtimeWinsTieBreak(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(205)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 2, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "config"
	def := configTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for _, leafID := range sim.LeafIDs() {
		upsertRow(t, sim.Leaf(leafID).Repo(), table, map[string]any{
			"key":   "lang",
			"value": "en",
		}, 100)
	}

	runAndCheck(t, sim, sev, seed, stepsFor(200))

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
		assertRowCount(t, masterRepo, "master", table, def, 1)
		for _, leafID := range sim.LeafIDs() {
			assertRowCount(t, sim.Leaf(leafID).Repo(), string(leafID), table, def, 1)
		}
	}
}

func TestTS_SelfWriteEnforcement(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(206)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "peer_status"
	def := selfWriteTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i, leafID := range sim.LeafIDs() {
		upsertRow(t, sim.Leaf(leafID).Repo(), table, map[string]any{
			"peer_id":   fmt.Sprintf("peer-%d", i),
			"last_seen": int64(1000 + i*10),
			"health":    "healthy",
		}, int64(100+i))
	}

	runAndCheck(t, sim, sev, seed, stepsFor(300))

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
		assertRowCount(t, masterRepo, "master", table, def, 3)
		for _, leafID := range sim.LeafIDs() {
			assertRowCount(t, sim.Leaf(leafID).Repo(), string(leafID), table, def, 3)
		}
	} else {
		logRowCounts(t, sim, table, def)
	}
}

// =============================================================================
// Level 2: Adversarial (10% faults)
// =============================================================================

func TestTS_PartitionHeal(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(300)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "devices"
	def := deviceTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	upsertRow(t, masterRepo, table, map[string]any{
		"device_id": "shared", "hostname": "base", "status": "ok",
	}, 100)
	if err := sim.Run(stepsFor(50)); err != nil {
		t.Fatalf("Run phase 1: %v", err)
	}

	sim.Network().Partition("leaf-0")
	upsertRow(t, sim.Leaf("leaf-0").Repo(), table, map[string]any{
		"device_id": "isolated", "hostname": "from-0", "status": "partitioned",
	}, 200)
	upsertRow(t, sim.Leaf("leaf-1").Repo(), table, map[string]any{
		"device_id": "connected", "hostname": "from-1", "status": "ok",
	}, 200)
	if err := sim.Run(stepsFor(100)); err != nil {
		t.Fatalf("Run phase 2: %v", err)
	}

	sim.Network().Heal("leaf-0")
	for _, leafID := range sim.LeafIDs() {
		sim.ScheduleSyncNow(leafID)
	}
	if err := sim.Run(stepsFor(150)); err != nil {
		t.Fatalf("Run phase 3: %v", err)
	}

	if err := sim.CheckSafety(); err != nil {
		t.Fatalf("Safety: %v", err)
	}
	t.Logf("[%s] seed=%d steps=%d syncs=%d errors=%d",
		sev.Name, seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
	} else {
		logRowCounts(t, sim, table, def)
	}
}

func TestTS_PartitionConcurrentSamePK(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(301)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 2, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "config"
	def := configTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	sim.Network().Partition("leaf-0")

	upsertRow(t, sim.Leaf("leaf-0").Repo(), table, map[string]any{
		"key": "mode", "value": "old",
	}, 100)
	upsertRow(t, sim.Leaf("leaf-1").Repo(), table, map[string]any{
		"key": "mode", "value": "new",
	}, 200)
	if err := sim.Run(stepsFor(100)); err != nil {
		t.Fatalf("Run partitioned: %v", err)
	}

	sim.Network().Heal("leaf-0")
	for _, leafID := range sim.LeafIDs() {
		sim.ScheduleSyncNow(leafID)
	}
	if err := sim.Run(stepsFor(200)); err != nil {
		t.Fatalf("Run healed: %v", err)
	}

	if err := sim.CheckSafety(); err != nil {
		t.Fatalf("Safety: %v", err)
	}
	t.Logf("[%s] seed=%d steps=%d syncs=%d errors=%d",
		sev.Name, seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
		pk := map[string]any{"key": "mode"}
		assertRowValue(t, masterRepo, "master", table, def, pk, "value", "new")
	}
}

func TestTS_BuggifyConvergence100(t *testing.T) {
	seed := seedFor(302)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, true)
	sim.Network().SetDropRate(0.10)

	table := "sensors"
	def := repo.TableDef{
		Columns: []repo.ColumnDef{
			{Name: "sensor_id", Type: "integer", PK: true},
			{Name: "reading", Type: "real"},
		},
		Conflict: "mtime-wins",
	}
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i := 0; i < 100; i++ {
		upsertRow(t, masterRepo, table, map[string]any{
			"sensor_id": int64(i),
			"reading":   float64(20.0 + float64(i)*0.1),
		}, 100)
	}

	if err := sim.Run(stepsFor(500)); err != nil {
		t.Fatalf("Run: %v", err)
	}

	t.Logf("[buggify] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	for _, leafID := range sim.LeafIDs() {
		r := sim.Leaf(leafID).Repo()
		if err := CheckDeltaChains(string(leafID), r); err != nil {
			t.Fatalf("Delta chain: %v", err)
		}
		if err := CheckNoOrphanPhantoms(string(leafID), r); err != nil {
			t.Fatalf("Orphan phantom: %v", err)
		}
	}

	logRowCounts(t, sim, table, def)
}

func TestTS_BuggifyMultiTable(t *testing.T) {
	seed := seedFor(303)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, true)
	sim.Network().SetDropRate(0.10)

	names := []string{"alpha", "beta", "gamma"}
	def := configTableDef()
	for _, name := range names {
		registerTableAll(t, sim, masterRepo, name, def, 100)
		for j := 0; j < 10; j++ {
			upsertRow(t, masterRepo, name, map[string]any{
				"key":   fmt.Sprintf("%s-k%d", name, j),
				"value": fmt.Sprintf("%s-v%d", name, j),
			}, 100)
		}
	}

	if err := sim.Run(stepsFor(500)); err != nil {
		t.Fatalf("Run: %v", err)
	}

	t.Logf("[buggify-multi] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	for _, leafID := range sim.LeafIDs() {
		r := sim.Leaf(leafID).Repo()
		if err := CheckDeltaChains(string(leafID), r); err != nil {
			t.Fatalf("Delta chain: %v", err)
		}
	}

	for _, name := range names {
		t.Logf("  table %s:", name)
		logRowCounts(t, sim, name, def)
	}
}

func TestTS_StalePeerRejoin(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(304)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "devices"
	def := deviceTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i := 0; i < 5; i++ {
		upsertRow(t, masterRepo, table, map[string]any{
			"device_id": fmt.Sprintf("init-%d", i),
			"hostname":  fmt.Sprintf("host-%d", i),
			"status":    "active",
		}, 100)
	}
	if err := sim.Run(stepsFor(50)); err != nil {
		t.Fatalf("Run phase 1: %v", err)
	}

	sim.Network().Partition("leaf-2")
	for i := 0; i < 5; i++ {
		upsertRow(t, masterRepo, table, map[string]any{
			"device_id": fmt.Sprintf("new-%d", i),
			"hostname":  fmt.Sprintf("new-host-%d", i),
			"status":    "pending",
		}, 200)
	}
	if err := sim.Run(stepsFor(100)); err != nil {
		t.Fatalf("Run phase 2: %v", err)
	}

	sim.Network().Heal("leaf-2")
	sim.ScheduleSyncNow("leaf-2")
	if err := sim.Run(stepsFor(150)); err != nil {
		t.Fatalf("Run phase 3: %v", err)
	}

	if err := sim.CheckSafety(); err != nil {
		t.Fatalf("Safety: %v", err)
	}
	t.Logf("[%s] seed=%d steps=%d syncs=%d errors=%d",
		sev.Name, seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
		assertRowCount(t, sim.Leaf("leaf-2").Repo(), "leaf-2", table, def, 10)
	} else {
		logRowCounts(t, sim, table, def)
	}
}

func TestTS_SchemaBeforeRows(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(305)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 2, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "metrics"
	def := configTableDef()
	registerTable(t, masterRepo, table, def, 100)

	if err := sim.Run(stepsFor(100)); err != nil {
		t.Fatalf("Run phase 1: %v", err)
	}

	for i := 0; i < 5; i++ {
		upsertRow(t, masterRepo, table, map[string]any{
			"key":   fmt.Sprintf("metric-%d", i),
			"value": fmt.Sprintf("%d", i*100),
		}, 200)
	}
	if err := sim.Run(stepsFor(200)); err != nil {
		t.Fatalf("Run phase 2: %v", err)
	}

	if err := sim.CheckSafety(); err != nil {
		t.Fatalf("Safety: %v", err)
	}
	t.Logf("[%s] seed=%d steps=%d syncs=%d errors=%d",
		sev.Name, seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	if isNormalMode(sev) {
		if err := sim.CheckAllTableSyncConverged(); err != nil {
			t.Fatalf("TableSync convergence: %v", err)
		}
		for _, leafID := range sim.LeafIDs() {
			assertRowCount(t, sim.Leaf(leafID).Repo(), string(leafID), table, def, 5)
		}
	} else {
		logRowCounts(t, sim, table, def)
	}
}

func TestTS_OwnerWriteEnforcement(t *testing.T) {
	sev := parseSeverity()
	seed := seedFor(306)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, sev.Buggify)
	sim.Network().SetDropRate(sev.DropRate)

	table := "resources"
	def := ownerWriteTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i, leafID := range sim.LeafIDs() {
		upsertRow(t, sim.Leaf(leafID).Repo(), table, map[string]any{
			"resource_id": fmt.Sprintf("res-%d", i),
			"data":        fmt.Sprintf("data-from-%d", i),
			"_owner":      string(leafID),
		}, int64(100+i))
	}

	if err := sim.Run(stepsFor(300)); err != nil {
		t.Fatalf("Run: %v", err)
	}

	if err := sim.CheckSafety(); err != nil {
		t.Fatalf("Safety: %v", err)
	}
	t.Logf("[%s] seed=%d steps=%d syncs=%d errors=%d",
		sev.Name, seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	logRowCounts(t, sim, table, def)
}

// =============================================================================
// Level 3: Hostile (25% faults)
// =============================================================================

func newHostileSim(t *testing.T, seed int64, numLeaves int, uvEnabled bool) (*Simulator, *repo.Repo, *MockFossil) {
	t.Helper()
	masterRepo := createMasterRepo(t)
	mf := NewMockFossil(masterRepo)
	sim, err := New(SimConfig{
		Seed:         seed,
		NumLeaves:    numLeaves,
		PollInterval: 5 * time.Second,
		TmpDir:       t.TempDir(),
		Upstream:     mf,
		Buggify:      true,
		UV:           uvEnabled,
	})
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	t.Cleanup(func() { sim.Close() })
	sim.Network().SetDropRate(0.25)
	sim.SetMasterRepo(masterRepo)
	return sim, masterRepo, mf
}

func assertBoundedProgress(t *testing.T, sim *Simulator, table string, def repo.TableDef, minRows int) {
	t.Helper()
	for _, leafID := range sim.LeafIDs() {
		rows, _, err := repo.ListXRows(sim.Leaf(leafID).Repo().DB(), table, def)
		if err != nil {
			t.Fatalf("ListXRows %s: %v", leafID, err)
		}
		if len(rows) < minRows {
			t.Errorf("%s: expected >= %d rows, got %d", leafID, minRows, len(rows))
		}
	}
}

func checkStructuralIntegrity(t *testing.T, sim *Simulator) {
	t.Helper()
	for _, leafID := range sim.LeafIDs() {
		r := sim.Leaf(leafID).Repo()
		if err := CheckDeltaChains(string(leafID), r); err != nil {
			t.Fatalf("Delta chain %s: %v", leafID, err)
		}
		if err := CheckNoOrphanPhantoms(string(leafID), r); err != nil {
			t.Fatalf("Orphan phantom %s: %v", leafID, err)
		}
	}
}

func TestTS_HostileConvergence(t *testing.T) {
	seed := seedFor(400)
	sim, masterRepo, _ := newHostileSim(t, seed, 5, false)

	table := "devices"
	def := deviceTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i, leafID := range sim.LeafIDs() {
		r := sim.Leaf(leafID).Repo()
		for j := 0; j < 10; j++ {
			upsertRow(t, r, table, map[string]any{
				"device_id": fmt.Sprintf("leaf%d-dev%d", i, j),
				"hostname":  fmt.Sprintf("host-%d-%d", i, j),
				"status":    "online",
			}, int64(100+i*100+j))
		}
	}

	if err := sim.Run(stepsFor(300)); err != nil {
		t.Fatalf("Run: %v", err)
	}
	t.Logf("[hostile] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	checkStructuralIntegrity(t, sim)
	assertBoundedProgress(t, sim, table, def, 10)
	logRowCounts(t, sim, table, def)
}

func TestTS_CorruptXRowPayload(t *testing.T) {
	seed := seedFor(401)
	sim, masterRepo, _ := newHostileSim(t, seed, 3, false)

	table := "config"
	def := configTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i := 0; i < 20; i++ {
		upsertRow(t, masterRepo, table, map[string]any{
			"key":   fmt.Sprintf("key-%d", i),
			"value": fmt.Sprintf("val-%d", i),
		}, 100)
	}

	if err := sim.Run(stepsFor(500)); err != nil {
		t.Fatalf("Run: %v", err)
	}
	t.Logf("[corrupt-payload] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	checkStructuralIntegrity(t, sim)
	assertBoundedProgress(t, sim, table, def, 1)
	logRowCounts(t, sim, table, def)
}

func TestTS_DropSchemaCards(t *testing.T) {
	seed := seedFor(402)
	sim, masterRepo, _ := newHostileSim(t, seed, 3, false)

	table := "alerts"
	def := configTableDef()
	registerTable(t, masterRepo, table, def, 100)

	for i := 0; i < 10; i++ {
		upsertRow(t, masterRepo, table, map[string]any{
			"key":   fmt.Sprintf("alert-%d", i),
			"value": fmt.Sprintf("msg-%d", i),
		}, 100)
	}

	if err := sim.Run(stepsFor(500)); err != nil {
		t.Fatalf("Run: %v", err)
	}
	t.Logf("[drop-schema] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	checkStructuralIntegrity(t, sim)

	schemaDeployed := 0
	for _, leafID := range sim.LeafIDs() {
		tables, err := repo.ListSyncedTables(sim.Leaf(leafID).Repo().DB())
		if err != nil {
			t.Fatalf("ListSyncedTables %s: %v", leafID, err)
		}
		for _, tbl := range tables {
			if tbl.Name == table {
				schemaDeployed++
				break
			}
		}
	}
	t.Logf("  schema deployed to %d/%d leaves", schemaDeployed, len(sim.LeafIDs()))
	if schemaDeployed == 0 {
		t.Errorf("schema not deployed to any leaf after 500 steps")
	}
}

func TestTS_TruncatedXIGotList(t *testing.T) {
	seed := seedFor(403)
	sim, masterRepo, _ := newHostileSim(t, seed, 3, false)

	table := "sensors"
	def := configTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i, leafID := range sim.LeafIDs() {
		r := sim.Leaf(leafID).Repo()
		for j := 0; j < 10; j++ {
			upsertRow(t, r, table, map[string]any{
				"key":   fmt.Sprintf("s%d-%d", i, j),
				"value": fmt.Sprintf("reading-%d-%d", i, j),
			}, int64(100+i*10+j))
		}
	}

	if err := sim.Run(stepsFor(600)); err != nil {
		t.Fatalf("Run: %v", err)
	}
	t.Logf("[truncated-xigot] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	checkStructuralIntegrity(t, sim)
	assertBoundedProgress(t, sim, table, def, 10)
	logRowCounts(t, sim, table, def)
}

func TestTS_CatalogHashCorruption(t *testing.T) {
	seed := seedFor(404)
	sim, masterRepo, _ := newHostileSim(t, seed, 2, false)

	table := "config"
	def := configTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	allRepos := []*repo.Repo{masterRepo}
	for _, leafID := range sim.LeafIDs() {
		allRepos = append(allRepos, sim.Leaf(leafID).Repo())
	}
	for _, r := range allRepos {
		for i := 0; i < 15; i++ {
			upsertRow(t, r, table, map[string]any{
				"key":   fmt.Sprintf("k%d", i),
				"value": fmt.Sprintf("v%d", i),
			}, 100)
		}
	}

	if err := sim.Run(stepsFor(400)); err != nil {
		t.Fatalf("Run: %v", err)
	}
	t.Logf("[catalog-corrupt] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	checkStructuralIntegrity(t, sim)
	assertBoundedProgress(t, sim, table, def, 15)
	logRowCounts(t, sim, table, def)
}

func TestTS_MixedWorkload(t *testing.T) {
	seed := seedFor(405)
	sim, masterRepo, mf := newHostileSim(t, seed, 3, true)

	for i := 0; i < 30; i++ {
		mf.StoreArtifact([]byte(fmt.Sprintf("mixed-blob-%04d-seed%d", i, seed)))
	}

	uv.EnsureSchema(masterRepo.DB())
	for i := 0; i < 5; i++ {
		uv.Write(masterRepo.DB(), fmt.Sprintf("mixed/file-%d.txt", i),
			[]byte(fmt.Sprintf("mixed-uv-%d-seed%d", i, seed)), int64(100+i))
	}

	table := "telemetry"
	def := configTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)
	for i := 0; i < 20; i++ {
		upsertRow(t, masterRepo, table, map[string]any{
			"key":   fmt.Sprintf("metric-%d", i),
			"value": fmt.Sprintf("%d", i*100),
		}, 100)
	}

	if err := sim.Run(stepsFor(600)); err != nil {
		t.Fatalf("Run: %v", err)
	}
	t.Logf("[mixed-workload] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	checkStructuralIntegrity(t, sim)

	for _, leafID := range sim.LeafIDs() {
		c, _ := CountBlobs(sim.Leaf(leafID).Repo())
		t.Logf("  %s: %d blobs", leafID, c)
		if c == 0 {
			t.Errorf("%s: zero blobs after mixed workload", leafID)
		}
	}

	assertBoundedProgress(t, sim, table, def, 1)
	logRowCounts(t, sim, table, def)
}

// =============================================================================
// Adversarial deletion tests
// =============================================================================

func TestTS_Adversarial_DeletionConvergence(t *testing.T) {
	seed := seedFor(307)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 3, true)
	sim.Network().SetDropRate(0.10)

	def := deviceTableDef()
	registerTableAll(t, sim, masterRepo, "devices", def, 1000)

	for i := 0; i < 5; i++ {
		id := fmt.Sprintf("d%d", i)
		upsertRow(t, masterRepo, "devices", map[string]any{
			"device_id": id,
			"hostname":  fmt.Sprintf("host-%d", i),
			"status":    "online",
		}, 1000)
	}

	if err := sim.Run(stepsFor(300)); err != nil {
		t.Fatalf("Run (distribute): %v", err)
	}

	pkColDefs := []repo.ColumnDef{{Name: "device_id", Type: "text", PK: true}}
	for i := 0; i < 3; i++ {
		id := fmt.Sprintf("d%d", i)
		pkHash := repo.PKHash(pkColDefs, map[string]any{"device_id": id})
		if _, err := repo.DeleteXRowByPKHash(masterRepo.DB(), "devices", def, pkHash, 2000); err != nil {
			t.Fatalf("DeleteXRowByPKHash d%d: %v", i, err)
		}
	}

	if err := sim.Run(stepsFor(500)); err != nil {
		t.Fatalf("Run (converge): %v", err)
	}

	t.Logf("[adversarial-deletion] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	if err := sim.CheckSafety(); err != nil {
		t.Fatalf("Safety: %v", err)
	}

	for _, leafID := range sim.LeafIDs() {
		leafRepo := sim.Leaf(leafID).Repo()
		rows, _, err := repo.ListXRows(leafRepo.DB(), "devices", def)
		if err != nil {
			t.Fatalf("ListXRows %s: %v", leafID, err)
		}
		tombstones := 0
		live := 0
		for _, row := range rows {
			if repo.IsTombstone(def, row) {
				tombstones++
			} else {
				live++
			}
		}
		if tombstones != 3 {
			t.Errorf("%s: tombstones=%d, want 3", leafID, tombstones)
		}
		if live != 2 {
			t.Errorf("%s: live=%d, want 2", leafID, live)
		}
	}

	masterHash, err := repo.CatalogHash(masterRepo.DB(), "devices", def)
	if err != nil {
		t.Fatalf("CatalogHash master: %v", err)
	}
	for _, leafID := range sim.LeafIDs() {
		leafHash, err := repo.CatalogHash(sim.Leaf(leafID).Repo().DB(), "devices", def)
		if err != nil {
			t.Fatalf("CatalogHash %s: %v", leafID, err)
		}
		if leafHash != masterHash {
			t.Errorf("%s: catalog hash %q != master %q", leafID, leafHash, masterHash)
		}
	}
}

func TestTS_Adversarial_DeleteUpdateRace(t *testing.T) {
	seed := seedFor(308)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 2, true)
	sim.Network().SetDropRate(0.10)

	def := deviceTableDef()
	registerTableAll(t, sim, masterRepo, "devices", def, 1000)

	upsertRow(t, masterRepo, "devices", map[string]any{
		"device_id": "contested",
		"hostname":  "original",
		"status":    "online",
	}, 1000)
	if err := sim.Run(stepsFor(300)); err != nil {
		t.Fatalf("Run (seed): %v", err)
	}

	pkColDefs := []repo.ColumnDef{{Name: "device_id", Type: "text", PK: true}}
	pkHash := repo.PKHash(pkColDefs, map[string]any{"device_id": "contested"})
	if _, err := repo.DeleteXRowByPKHash(masterRepo.DB(), "devices", def, pkHash, 2000); err != nil {
		t.Fatalf("DeleteXRowByPKHash: %v", err)
	}
	leaf0 := sim.Leaf(sim.LeafIDs()[0]).Repo()
	upsertRow(t, leaf0, "devices", map[string]any{
		"device_id": "contested",
		"hostname":  "updated",
		"status":    "active",
	}, 3000)

	if err := sim.Run(stepsFor(500)); err != nil {
		t.Fatalf("Run (race): %v", err)
	}

	t.Logf("[adversarial-race] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	if err := sim.CheckSafety(); err != nil {
		t.Fatalf("Safety: %v", err)
	}

	for _, leafID := range sim.LeafIDs() {
		row, mtime, err := repo.LookupXRow(sim.Leaf(leafID).Repo().DB(), "devices", def, pkHash)
		if err != nil {
			t.Fatalf("LookupXRow %s: %v", leafID, err)
		}
		if row == nil {
			t.Errorf("%s: row missing", leafID)
			continue
		}
		if repo.IsTombstone(def, row) {
			t.Errorf("%s: should be live (mtime=3000 wins over delete at mtime=2000)", leafID)
		}
		if mtime != 3000 {
			t.Errorf("%s: mtime=%d, want 3000", leafID, mtime)
		}
	}
	row, mtime, err := repo.LookupXRow(masterRepo.DB(), "devices", def, pkHash)
	if err != nil {
		t.Fatalf("LookupXRow master: %v", err)
	}
	if row == nil || repo.IsTombstone(def, row) || mtime != 3000 {
		t.Errorf("master: row=%v tombstone=%v mtime=%d, want live row at mtime=3000",
			row != nil, row != nil && repo.IsTombstone(def, row), mtime)
	}
}

func TestTS_Adversarial_DeleteUnseenRow(t *testing.T) {
	seed := seedFor(309)
	sim, masterRepo, _ := newTableSyncSim(t, seed, 2, true)
	sim.Network().SetDropRate(0.10)

	def := deviceTableDef()
	registerTableAll(t, sim, masterRepo, "devices", def, 1000)

	upsertRow(t, masterRepo, "devices", map[string]any{
		"device_id": "ephemeral",
		"hostname":  "ghost",
		"status":    "gone",
	}, 1000)
	pkColDefs := []repo.ColumnDef{{Name: "device_id", Type: "text", PK: true}}
	pkHash := repo.PKHash(pkColDefs, map[string]any{"device_id": "ephemeral"})
	if _, err := repo.DeleteXRowByPKHash(masterRepo.DB(), "devices", def, pkHash, 2000); err != nil {
		t.Fatalf("DeleteXRowByPKHash: %v", err)
	}

	if err := sim.Run(stepsFor(400)); err != nil {
		t.Fatalf("Run: %v", err)
	}

	t.Logf("[adversarial-unseen] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	if err := sim.CheckSafety(); err != nil {
		t.Fatalf("Safety: %v", err)
	}

	for _, leafID := range sim.LeafIDs() {
		row, _, err := repo.LookupXRow(sim.Leaf(leafID).Repo().DB(), "devices", def, pkHash)
		if err != nil {
			t.Fatalf("LookupXRow %s: %v", leafID, err)
		}
		if row == nil {
			t.Errorf("%s: should have tombstone for unseen row", leafID)
			continue
		}
		if !repo.IsTombstone(def, row) {
			t.Errorf("%s: should be tombstone, got live row", leafID)
		}
	}
}

func TestTS_StressTest1000Rows(t *testing.T) {
	seed := seedFor(406)
	sim, masterRepo, _ := newHostileSim(t, seed, 2, false)

	table := "events"
	def := configTableDef()
	registerTableAll(t, sim, masterRepo, table, def, 100)

	for i := 0; i < 500; i++ {
		upsertRow(t, masterRepo, table, map[string]any{
			"key":   fmt.Sprintf("evt-%04d", i),
			"value": fmt.Sprintf("data-%04d", i),
		}, int64(100+i))
	}

	if err := sim.Run(stepsFor(50)); err != nil {
		t.Fatalf("Run: %v", err)
	}
	t.Logf("[stress-1000] seed=%d steps=%d syncs=%d errors=%d",
		seed, sim.Steps, sim.TotalSyncs, sim.TotalErrors)

	checkStructuralIntegrity(t, sim)
	assertBoundedProgress(t, sim, table, def, 1)
	logRowCounts(t, sim, table, def)
}

// =============================================================================
// Deletion, Resurrection, and Integer PK tests
// =============================================================================

func TestTableSync_Deletion_Convergence(t *testing.T) {
	sim, masterRepo, _ := newTableSyncSim(t, 42, 3, false)
	def := deviceTableDef()
	registerTableAll(t, sim, masterRepo, "devices", def, 1000)

	upsertRow(t, masterRepo, "devices", map[string]any{
		"device_id": "d1", "hostname": "alpha", "status": "online",
	}, 1000)

	if err := sim.Run(30); err != nil {
		t.Fatalf("Run (seed): %v", err)
	}
	for _, leafID := range sim.LeafIDs() {
		assertRowCount(t, sim.Leaf(leafID).Repo(), string(leafID), "devices", def, 1)
	}

	pkColDefs := []repo.ColumnDef{{Name: "device_id", Type: "text", PK: true}}
	pkHash := repo.PKHash(pkColDefs, map[string]any{"device_id": "d1"})
	deleted, err := repo.DeleteXRowByPKHash(masterRepo.DB(), "devices", def, pkHash, 2000)
	if err != nil || !deleted {
		t.Fatalf("master delete: deleted=%v err=%v", deleted, err)
	}

	if err := sim.Run(50); err != nil {
		t.Fatalf("Run (delete): %v", err)
	}

	for _, leafID := range sim.LeafIDs() {
		row, mtime, _ := repo.LookupXRow(sim.Leaf(leafID).Repo().DB(), "devices", def, pkHash)
		if row == nil {
			t.Errorf("%s: row should exist as tombstone", leafID)
			continue
		}
		if !repo.IsTombstone(def, row) {
			t.Errorf("%s: row should be tombstone", leafID)
		}
		if mtime != 2000 {
			t.Errorf("%s: mtime=%d, want 2000", leafID, mtime)
		}
	}
}

func TestTableSync_Deletion_Resurrection(t *testing.T) {
	sim, masterRepo, _ := newTableSyncSim(t, 99, 2, false)
	def := deviceTableDef()
	registerTableAll(t, sim, masterRepo, "devices", def, 1000)

	upsertRow(t, masterRepo, "devices", map[string]any{
		"device_id": "d1", "hostname": "alpha", "status": "online",
	}, 1000)
	if err := sim.Run(30); err != nil {
		t.Fatalf("Run (seed): %v", err)
	}

	pkColDefs := []repo.ColumnDef{{Name: "device_id", Type: "text", PK: true}}
	pkHash := repo.PKHash(pkColDefs, map[string]any{"device_id": "d1"})
	repo.DeleteXRowByPKHash(masterRepo.DB(), "devices", def, pkHash, 2000)

	if err := sim.Run(30); err != nil {
		t.Fatalf("Run (delete): %v", err)
	}

	leaf0 := sim.Leaf(sim.LeafIDs()[0]).Repo()
	upsertRow(t, leaf0, "devices", map[string]any{
		"device_id": "d1", "hostname": "beta", "status": "active",
	}, 3000)

	if err := sim.Run(50); err != nil {
		t.Fatalf("Run (resurrect): %v", err)
	}

	for _, leafID := range sim.LeafIDs() {
		row, mtime, _ := repo.LookupXRow(sim.Leaf(leafID).Repo().DB(), "devices", def, pkHash)
		if row == nil {
			t.Errorf("%s: row missing after resurrection", leafID)
			continue
		}
		if repo.IsTombstone(def, row) {
			t.Errorf("%s: row should be live after resurrection", leafID)
		}
		if mtime != 3000 {
			t.Errorf("%s: mtime=%d, want 3000", leafID, mtime)
		}
	}
	row, mtime, _ := repo.LookupXRow(masterRepo.DB(), "devices", def, pkHash)
	if row == nil || repo.IsTombstone(def, row) || mtime != 3000 {
		t.Errorf("master: row=%v tombstone=%v mtime=%d", row != nil, row != nil && repo.IsTombstone(def, row), mtime)
	}
}

func TestTableSync_IntegerPK_Convergence(t *testing.T) {
	sim, masterRepo, _ := newTableSyncSim(t, 77, 2, false)
	def := repo.TableDef{
		Columns: []repo.ColumnDef{
			{Name: "seq", Type: "integer", PK: true},
			{Name: "payload", Type: "text"},
		},
		Conflict: "mtime-wins",
	}
	registerTableAll(t, sim, masterRepo, "events", def, 1000)

	bigPK := int64(1<<53 + 1)
	upsertRow(t, masterRepo, "events", map[string]any{
		"seq": bigPK, "payload": "event-data",
	}, 1000)

	if err := sim.Run(100); err != nil {
		t.Fatalf("Run: %v", err)
	}

	masterHash, _ := repo.CatalogHash(masterRepo.DB(), "events", def)
	for _, leafID := range sim.LeafIDs() {
		leafHash, _ := repo.CatalogHash(sim.Leaf(leafID).Repo().DB(), "events", def)
		if leafHash != masterHash {
			t.Errorf("%s: catalog hash %q != master %q", leafID, leafHash, masterHash)
		}
	}
}

func TestTableSync_Deletion_CompositePK(t *testing.T) {
	sim, masterRepo, _ := newTableSyncSim(t, 55, 2, false)
	def := repo.TableDef{
		Columns: []repo.ColumnDef{
			{Name: "org", Type: "text", PK: true},
			{Name: "user_id", Type: "text", PK: true},
			{Name: "role", Type: "text"},
		},
		Conflict: "mtime-wins",
	}
	registerTableAll(t, sim, masterRepo, "members", def, 1000)

	upsertRow(t, masterRepo, "members", map[string]any{
		"org": "acme", "user_id": "alice", "role": "admin",
	}, 1000)
	upsertRow(t, masterRepo, "members", map[string]any{
		"org": "acme", "user_id": "bob", "role": "member",
	}, 1000)

	if err := sim.Run(50); err != nil {
		t.Fatalf("Run (seed): %v", err)
	}

	pkColDefs := []repo.ColumnDef{
		{Name: "org", Type: "text", PK: true},
		{Name: "user_id", Type: "text", PK: true},
	}
	aliceHash := repo.PKHash(pkColDefs, map[string]any{"org": "acme", "user_id": "alice"})
	if _, err := repo.DeleteXRowByPKHash(masterRepo.DB(), "members", def, aliceHash, 2000); err != nil {
		t.Fatalf("DeleteXRowByPKHash alice: %v", err)
	}

	if err := sim.Run(50); err != nil {
		t.Fatalf("Run (delete): %v", err)
	}

	bobHash := repo.PKHash(pkColDefs, map[string]any{"org": "acme", "user_id": "bob"})
	for _, leafID := range sim.LeafIDs() {
		d := sim.Leaf(leafID).Repo().DB()
		aliceRow, _, _ := repo.LookupXRow(d, "members", def, aliceHash)
		if aliceRow == nil || !repo.IsTombstone(def, aliceRow) {
			t.Errorf("%s: alice should be tombstone", leafID)
		}
		bobRow, _, _ := repo.LookupXRow(d, "members", def, bobHash)
		if bobRow == nil || repo.IsTombstone(def, bobRow) {
			t.Errorf("%s: bob should be live", leafID)
		}
	}
	aliceRow, _, _ := repo.LookupXRow(masterRepo.DB(), "members", def, aliceHash)
	if aliceRow == nil || !repo.IsTombstone(def, aliceRow) {
		t.Error("master: alice should be tombstone")
	}
	bobRow, _, _ := repo.LookupXRow(masterRepo.DB(), "members", def, bobHash)
	if bobRow == nil || repo.IsTombstone(def, bobRow) {
		t.Error("master: bob should be live")
	}
}