package persistence

import (
	"context"
	"database/sql"
	"errors"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/nexuscrm/db"
)

func TestHistorySequenceTiDBOrderingAtomicityAndIsolation(t *testing.T) {
	rawDB, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	rawDB.SetMaxOpenConns(8)
	ctx := context.Background()
	for _, ddl := range []string{
		`CREATE TABLE _System_Field (
			Id VARCHAR(255) PRIMARY KEY,
			TrackHistory TINYINT(1) NOT NULL DEFAULT 0,
			HistoryTrackingRevision BIGINT NOT NULL DEFAULT 0,
			IsCustomized TINYINT(1) NOT NULL DEFAULT 0,
			IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
			LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
		)`,
		`CREATE TABLE _System_FieldHistorySequence (
			FieldId VARCHAR(255) PRIMARY KEY,
			CurrentSequence BIGINT NOT NULL DEFAULT 0,
			CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
			LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
		)`,
		`CREATE TABLE _System_FieldHistoryTrackingInterval (
			Id VARCHAR(255) PRIMARY KEY,
			FieldId VARCHAR(255) NOT NULL,
			OpenFieldId VARCHAR(255) NULL,
			StartedSequence BIGINT NOT NULL,
			EndedSequence BIGINT NULL,
			StartedAt DATETIME(6) NOT NULL,
			EndedAt DATETIME(6) NULL,
			UNIQUE KEY uk_field_history_open (OpenFieldId)
		)`,
		`CREATE TABLE _System_AuditLog (
			Id VARCHAR(255) PRIMARY KEY,
			FieldId VARCHAR(255) NOT NULL,
			HistorySequence BIGINT NULL,
			EventAt DATETIME(6) NULL
		)`,
		`CREATE TABLE BusinessRecord (
			Id VARCHAR(255) PRIMARY KEY,
			Value VARCHAR(255) NOT NULL
		)`,
	} {
		if _, err := rawDB.ExecContext(ctx, ddl); err != nil {
			t.Fatalf("create sequence integration table: %v", err)
		}
	}
	if _, err := rawDB.Exec(`INSERT INTO _System_Field (Id) VALUES ('field-a'), ('field-b'), ('field-c'), ('field-d'), ('field-e'), ('field-f'), ('field-g'), ('field-h')`); err != nil {
		t.Fatalf("seed fields: %v", err)
	}
	if _, err := rawDB.Exec(`INSERT INTO BusinessRecord (Id, Value) VALUES ('record-atomic', 'before')`); err != nil {
		t.Fatalf("seed business record: %v", err)
	}

	tm := NewTransactionManager(database.NewConnectionFromDB(rawDB))
	repo := NewSchemaRepository(rawDB, nil)
	reserveFact := func(id, fieldID string) error {
		return tm.WithRetry(ctx, func(txCtx context.Context) error {
			reserved, err := ReserveHistorySequences(txCtx, tm.ExtractTx(txCtx), []string{fieldID})
			if err != nil {
				return err
			}
			_, err = tm.ExtractTx(txCtx).(Executor).ExecContext(txCtx,
				`INSERT INTO _System_AuditLog (Id, FieldId, HistorySequence, EventAt) VALUES (?, ?, ?, '2026-07-17 00:00:00.000000')`,
				id, fieldID, reserved[fieldID])
			return err
		}, 3)
	}
	transitionField := func(fieldID string, desired bool) error {
		return tm.WithTransaction(ctx, func(txCtx context.Context) error {
			_, err := repo.ApplyHistoryTrackingTransition(txCtx, fieldID, desired, tm.ExtractTx(txCtx), false, false)
			return err
		})
	}
	transition := func(desired bool) error { return transitionField("field-a", desired) }
	eligible := func(id string) int {
		var count int
		if err := rawDB.QueryRow(`
			SELECT COUNT(*)
			FROM _System_AuditLog fact
			JOIN _System_FieldHistoryTrackingInterval interval_row ON interval_row.FieldId = fact.FieldId
			WHERE fact.Id = ?
			  AND fact.HistorySequence IS NOT NULL
			  AND fact.HistorySequence > interval_row.StartedSequence
			  AND (interval_row.EndedSequence IS NULL OR fact.HistorySequence < interval_row.EndedSequence)`, id).Scan(&count); err != nil {
			t.Fatalf("query eligibility for %s: %v", id, err)
		}
		return count
	}

	if err := reserveFact("before-enable", "field-a"); err != nil {
		t.Fatalf("reserve fact before enable: %v", err)
	}
	if err := transition(true); err != nil {
		t.Fatalf("enable after fact: %v", err)
	}
	if got := eligible("before-enable"); got != 0 {
		t.Fatalf("fact serialized before enable eligible = %d, want 0", got)
	}
	if err := reserveFact("after-enable", "field-a"); err != nil {
		t.Fatalf("reserve fact after enable: %v", err)
	}
	if got := eligible("after-enable"); got != 1 {
		t.Fatalf("fact serialized after enable eligible = %d, want 1", got)
	}
	if err := transition(false); err != nil {
		t.Fatalf("disable after fact: %v", err)
	}
	if got := eligible("after-enable"); got != 1 {
		t.Fatalf("fact serialized before disable eligible = %d, want 1", got)
	}
	if err := reserveFact("after-disable", "field-a"); err != nil {
		t.Fatalf("reserve fact after disable: %v", err)
	}
	if got := eligible("after-disable"); got != 0 {
		t.Fatalf("fact serialized after disable eligible = %d, want 0", got)
	}

	waitStarted := func(ch <-chan struct{}, owner string) {
		t.Helper()
		select {
		case <-ch:
		case <-time.After(10 * time.Second):
			t.Fatalf("timed out waiting for %s transaction start", owner)
		}
	}
	waitResult := func(ch <-chan error, owner string) {
		t.Helper()
		if err := <-ch; err != nil {
			t.Fatalf("%s transaction: %v", owner, err)
		}
	}
	pausedFact := func(id, fieldID string, started chan<- struct{}, proceed <-chan struct{}) <-chan error {
		result := make(chan error, 1)
		go func() {
			result <- tm.WithTransaction(ctx, func(txCtx context.Context) error {
				close(started)
				<-proceed
				reserved, err := ReserveHistorySequences(txCtx, tm.ExtractTx(txCtx), []string{fieldID})
				if err != nil {
					return err
				}
				_, err = tm.ExtractTx(txCtx).(Executor).ExecContext(txCtx,
					`INSERT INTO _System_AuditLog (Id, FieldId, HistorySequence, EventAt) VALUES (?, ?, ?, '2026-07-17 00:00:00.000000')`,
					id, fieldID, reserved[fieldID])
				return err
			})
		}()
		return result
	}
	pausedTransition := func(fieldID string, desired bool, started chan<- struct{}, proceed <-chan struct{}) <-chan error {
		result := make(chan error, 1)
		go func() {
			result <- tm.WithTransaction(ctx, func(txCtx context.Context) error {
				close(started)
				<-proceed
				_, err := repo.ApplyHistoryTrackingTransition(txCtx, fieldID, desired, tm.ExtractTx(txCtx), false, false)
				return err
			})
		}()
		return result
	}
	resetRaceField := func(fieldID string, tracked bool) {
		t.Helper()
		if _, err := rawDB.Exec(`DELETE FROM _System_AuditLog WHERE FieldId = ?`, fieldID); err != nil {
			t.Fatalf("reset %s audit facts: %v", fieldID, err)
		}
		if _, err := rawDB.Exec(`DELETE FROM _System_FieldHistoryTrackingInterval WHERE FieldId = ?`, fieldID); err != nil {
			t.Fatalf("reset %s intervals: %v", fieldID, err)
		}
		if _, err := rawDB.Exec(`DELETE FROM _System_FieldHistorySequence WHERE FieldId = ?`, fieldID); err != nil {
			t.Fatalf("reset %s sequence: %v", fieldID, err)
		}
		if _, err := rawDB.Exec(`UPDATE _System_Field SET TrackHistory = 0, HistoryTrackingRevision = 0 WHERE Id = ?`, fieldID); err != nil {
			t.Fatalf("reset %s field: %v", fieldID, err)
		}
		if tracked {
			if err := transitionField(fieldID, true); err != nil {
				t.Fatalf("initialize %s tracked: %v", fieldID, err)
			}
		}
	}

	// Transaction start time is deliberately inverted from sequence/commit order.
	// Eligibility follows the serialized field sequence, never wall clock time.
	resetRaceField("field-c", false)
	started, proceed := make(chan struct{}), make(chan struct{})
	factResult := pausedFact("record-starts-before-enable", "field-c", started, proceed)
	waitStarted(started, "record-before-enable")
	if err := transitionField("field-c", true); err != nil {
		t.Fatalf("enable while earlier record transaction is paused: %v", err)
	}
	close(proceed)
	waitResult(factResult, "record-after-enable")

	resetRaceField("field-d", false)
	started, proceed = make(chan struct{}), make(chan struct{})
	transitionResult := pausedTransition("field-d", true, started, proceed)
	waitStarted(started, "enable-before-record")
	if err := reserveFact("record-commits-before-enable", "field-d"); err != nil {
		t.Fatalf("record while earlier enable transaction is paused: %v", err)
	}
	close(proceed)
	waitResult(transitionResult, "enable-after-record")

	resetRaceField("field-e", true)
	started, proceed = make(chan struct{}), make(chan struct{})
	factResult = pausedFact("record-starts-before-disable", "field-e", started, proceed)
	waitStarted(started, "record-before-disable")
	if err := transitionField("field-e", false); err != nil {
		t.Fatalf("disable while earlier record transaction is paused: %v", err)
	}
	close(proceed)
	waitResult(factResult, "record-after-disable")

	resetRaceField("field-f", true)
	started, proceed = make(chan struct{}), make(chan struct{})
	transitionResult = pausedTransition("field-f", false, started, proceed)
	waitStarted(started, "disable-before-record")
	if err := reserveFact("record-commits-before-disable", "field-f"); err != nil {
		t.Fatalf("record while earlier disable transaction is paused: %v", err)
	}
	close(proceed)
	waitResult(transitionResult, "disable-after-record")

	if _, err := rawDB.Exec(`UPDATE _System_AuditLog SET EventAt = '2026-07-17 00:00:00.000000' WHERE FieldId IN ('field-c', 'field-d', 'field-e', 'field-f')`); err != nil {
		t.Fatalf("equalize diagnostic fact time: %v", err)
	}
	if _, err := rawDB.Exec(`UPDATE _System_FieldHistoryTrackingInterval SET StartedAt = '2026-07-17 00:00:00.000000', EndedAt = CASE WHEN EndedAt IS NULL THEN NULL ELSE '2026-07-17 00:00:00.000000' END WHERE FieldId IN ('field-c', 'field-d', 'field-e', 'field-f')`); err != nil {
		t.Fatalf("equalize diagnostic interval time: %v", err)
	}
	for id, want := range map[string]int{
		"record-starts-before-enable":   1,
		"record-commits-before-enable":  0,
		"record-starts-before-disable":  0,
		"record-commits-before-disable": 1,
	} {
		if got := eligible(id); got != want {
			t.Fatalf("opposite-order fact %s eligible = %d, want %d", id, got, want)
		}
	}

	resetRaceField("field-g", false)
	if err := transitionField("field-g", true); err != nil {
		t.Fatalf("enable equality field: %v", err)
	}
	if err := transitionField("field-g", false); err != nil {
		t.Fatalf("disable equality field: %v", err)
	}
	if _, err := rawDB.Exec(`INSERT INTO _System_AuditLog (Id, FieldId, HistorySequence, EventAt) VALUES ('equal-start', 'field-g', 1, UTC_TIMESTAMP(6)), ('equal-end', 'field-g', 2, UTC_TIMESTAMP(6)), ('null-sequence', 'field-g', NULL, UTC_TIMESTAMP(6))`); err != nil {
		t.Fatalf("insert equality facts: %v", err)
	}
	for _, id := range []string{"equal-start", "equal-end", "null-sequence"} {
		if got := eligible(id); got != 0 {
			t.Fatalf("strict boundary fact %s eligible = %d, want 0", id, got)
		}
	}

	beforeRollback := map[string]int64{}
	for _, fieldID := range []string{"field-a", "field-b"} {
		var current int64
		if err := rawDB.QueryRow(`SELECT CurrentSequence FROM _System_FieldHistorySequence WHERE FieldId = ?`, fieldID).Scan(&current); err != nil && !errors.Is(err, sql.ErrNoRows) {
			t.Fatalf("read %s sequence before rollback: %v", fieldID, err)
		}
		beforeRollback[fieldID] = current
	}
	rollbackErr := errors.New("rollback whole record")
	err := tm.WithTransaction(ctx, func(txCtx context.Context) error {
		exec := tm.ExtractTx(txCtx).(Executor)
		if _, err := exec.ExecContext(txCtx, `UPDATE BusinessRecord SET Value = 'after' WHERE Id = 'record-atomic'`); err != nil {
			return err
		}
		reserved, err := ReserveHistorySequences(txCtx, exec, []string{"field-a", "field-b"})
		if err != nil {
			return err
		}
		for _, fieldID := range []string{"field-a", "field-b"} {
			if _, err := exec.ExecContext(txCtx, `INSERT INTO _System_AuditLog (Id, FieldId, HistorySequence) VALUES (?, ?, ?)`, "rollback-"+fieldID, fieldID, reserved[fieldID]); err != nil {
				return err
			}
		}
		return rollbackErr
	})
	if !errors.Is(err, rollbackErr) {
		t.Fatalf("multi-field rollback error = %v", err)
	}
	var businessValue string
	if err := rawDB.QueryRow(`SELECT Value FROM BusinessRecord WHERE Id = 'record-atomic'`).Scan(&businessValue); err != nil || businessValue != "before" {
		t.Fatalf("business rollback value = %q, err = %v", businessValue, err)
	}
	for fieldID, old := range beforeRollback {
		var current int64
		err := rawDB.QueryRow(`SELECT CurrentSequence FROM _System_FieldHistorySequence WHERE FieldId = ?`, fieldID).Scan(&current)
		if old == 0 && errors.Is(err, sql.ErrNoRows) {
			continue
		}
		if err != nil {
			t.Fatalf("read %s sequence after rollback: %v", fieldID, err)
		}
		if current != old {
			t.Fatalf("%s sequence after rollback = %d, want %d", fieldID, current, old)
		}
	}
	var rollbackFacts int
	if err := rawDB.QueryRow(`SELECT COUNT(*) FROM _System_AuditLog WHERE Id IN ('rollback-field-a', 'rollback-field-b')`).Scan(&rollbackFacts); err != nil || rollbackFacts != 0 {
		t.Fatalf("rollback audit facts = %d, err = %v", rollbackFacts, err)
	}

	before := map[string]int64{}
	for _, fieldID := range []string{"field-a", "field-b"} {
		var current int64
		if err := rawDB.QueryRow(`SELECT CurrentSequence FROM _System_FieldHistorySequence WHERE FieldId = ?`, fieldID).Scan(&current); err != nil && !errors.Is(err, sql.ErrNoRows) {
			t.Fatalf("read %s sequence before isolation check: %v", fieldID, err)
		}
		before[fieldID] = current
	}
	var wg sync.WaitGroup
	errs := make(chan error, 2)
	for _, fieldID := range []string{"field-a", "field-b"} {
		wg.Add(1)
		go func(fieldID string) {
			defer wg.Done()
			errs <- tm.WithRetry(ctx, func(txCtx context.Context) error {
				_, err := ReserveHistorySequences(txCtx, tm.ExtractTx(txCtx), []string{fieldID})
				return err
			}, 3)
		}(fieldID)
	}
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("different-field reservation conflict: %v", err)
		}
	}
	for fieldID, old := range before {
		var current int64
		if err := rawDB.QueryRow(`SELECT CurrentSequence FROM _System_FieldHistorySequence WHERE FieldId = ?`, fieldID).Scan(&current); err != nil {
			t.Fatalf("read %s sequence: %v", fieldID, err)
		}
		if current != old+1 {
			t.Fatalf("%s sequence = %d, want %d", fieldID, current, old+1)
		}
	}

	barrier := newHistoryCASBarrier(2)
	var attempts [2]atomic.Int32
	sequences := make([]int64, 2)
	errList := make([]error, 2)
	start := make(chan struct{})
	wg = sync.WaitGroup{}
	for i := range errList {
		wg.Add(1)
		go func(index int) {
			defer wg.Done()
			<-start
			errList[index] = tm.WithRetry(ctx, func(txCtx context.Context) error {
				attempts[index].Add(1)
				tx := tm.ExtractTx(txCtx).(*sql.Tx)
				reserved, err := ReserveHistorySequences(txCtx, &historyBarrierExecutor{Executor: tx, barrier: barrier}, []string{"field-h"})
				if err != nil {
					return err
				}
				sequences[index] = reserved["field-h"]
				_, err = tx.ExecContext(txCtx, `INSERT INTO _System_AuditLog (Id, FieldId, HistorySequence) VALUES (?, 'field-h', ?)`, "same-field-"+string(rune('a'+index)), sequences[index])
				return err
			}, 3)
		}(i)
	}
	close(start)
	wg.Wait()
	for i, err := range errList {
		if err != nil {
			t.Fatalf("same-field record %d after bounded retry: %v", i, err)
		}
	}
	if sequences[0] == sequences[1] {
		t.Fatalf("same-field records reused sequence %d", sequences[0])
	}
	if got := attempts[0].Load() + attempts[1].Load(); got < 2 || got > 3 {
		t.Fatalf("same-field total attempts = %d, want one attempt each and at most one typed root retry", got)
	}
}

func TestHistorySequenceTiDBCrossCallOppositeOrderRetriesWithCompletePrelock(t *testing.T) {
	rawDB, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	rawDB.SetMaxOpenConns(8)

	if _, err := rawDB.Exec(`CREATE TABLE _System_FieldHistorySequence (
		FieldId VARCHAR(255) PRIMARY KEY,
		CurrentSequence BIGINT NOT NULL DEFAULT 0,
		CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
		LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
	)`); err != nil {
		t.Fatalf("create sequence table: %v", err)
	}
	if _, err := rawDB.Exec(`INSERT INTO _System_FieldHistorySequence (FieldId, CurrentSequence)
		VALUES ('field-a', 0), ('field-z', 0)`); err != nil {
		t.Fatalf("seed sequence rows: %v", err)
	}

	tm := NewTransactionManager(database.NewConnectionFromDB(rawDB))
	tm.SetPessimisticLockWaitTimeout(5)
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()

	orders := [2][2]string{{"field-z", "field-a"}, {"field-a", "field-z"}}
	firstLocks := make(chan struct{}, len(orders))
	releaseSecond := make(chan struct{})
	var attempts [2]atomic.Int32
	errs := make([]error, len(orders))

	var wg sync.WaitGroup
	for i := range orders {
		wg.Add(1)
		go func(index int) {
			defer wg.Done()
			errs[index] = tm.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
				attempt := attempts[index].Add(1)
				for position, fieldID := range orders[index] {
					if _, err := ReserveHistorySequences(txCtx, tm.ExtractTx(txCtx), []string{fieldID}); err != nil {
						return err
					}
					if attempt == 1 && position == 0 {
						firstLocks <- struct{}{}
						select {
						case <-releaseSecond:
						case <-ctx.Done():
							return ctx.Err()
						}
					}
				}
				return nil
			}, 3)
		}(i)
	}

	for range orders {
		select {
		case <-firstLocks:
		case <-ctx.Done():
			t.Fatalf("wait for opposite-order first locks: %v", ctx.Err())
		}
	}
	close(releaseSecond)
	wg.Wait()

	for i, err := range errs {
		if err != nil {
			t.Fatalf("opposite-order transaction %d: %v", i, err)
		}
	}
	if got := attempts[0].Load(); got != 2 {
		t.Fatalf("descending transaction attempts = %d, want one planned retry", got)
	}
	if got := attempts[1].Load(); got != 1 {
		t.Fatalf("ascending transaction attempts = %d, want one attempt", got)
	}
	for _, fieldID := range []string{"field-a", "field-z"} {
		var current int64
		if err := rawDB.QueryRow(`SELECT CurrentSequence FROM _System_FieldHistorySequence WHERE FieldId = ?`, fieldID).Scan(&current); err != nil {
			t.Fatalf("read %s sequence: %v", fieldID, err)
		}
		if current != 2 {
			t.Fatalf("%s sequence = %d, want exactly two committed reservations and no prelock increment", fieldID, current)
		}
	}
}

func TestHistorySequenceTiDBLeadFirstWorkAndHistoryPrelockRetryConvergeWithoutDeadlock(t *testing.T) {
	rawDB, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	rawDB.SetMaxOpenConns(8)

	for _, ddl := range []string{
		"CREATE TABLE `Lead` (\n\t\t\tId VARCHAR(255) PRIMARY KEY,\n\t\t\tName VARCHAR(255) NOT NULL\n\t\t)",
		`CREATE TABLE _System_FieldHistorySequence (
			FieldId VARCHAR(255) PRIMARY KEY,
			CurrentSequence BIGINT NOT NULL DEFAULT 0,
			CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
			LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
		)`,
	} {
		if _, err := rawDB.Exec(ddl); err != nil {
			t.Fatalf("create retry ordering test table: %v", err)
		}
	}
	if _, err := rawDB.Exec("INSERT INTO `Lead` (Id, Name) VALUES ('lead-001', 'Lead')"); err != nil {
		t.Fatalf("seed Lead: %v", err)
	}
	if _, err := rawDB.Exec(`INSERT INTO _System_FieldHistorySequence (FieldId, CurrentSequence)
		VALUES ('field-a', 0), ('field-z', 0)`); err != nil {
		t.Fatalf("seed history sequences: %v", err)
	}

	tm := NewTransactionManager(database.NewConnectionFromDB(rawDB))
	tm.SetPessimisticLockWaitTimeout(5)
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()

	await := func(ch <-chan struct{}, name string) {
		t.Helper()
		select {
		case <-ch:
		case <-ctx.Done():
			t.Fatalf("timed out waiting for %s: %v", name, ctx.Err())
		}
	}

	retryFirstHistoryLocked := make(chan struct{})
	allowOutOfOrder := make(chan struct{})
	retrySecondStarted := make(chan struct{})
	normalLeadLocked := make(chan struct{})
	var retryAttempts atomic.Int32
	var normalAttempts atomic.Int32
	retryErr := make(chan error, 1)
	normalErr := make(chan error, 1)

	go func() {
		retryErr <- tm.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
			attempt := retryAttempts.Add(1)
			tx := tm.ExtractTx(txCtx).(Executor)
			if attempt == 1 {
				if _, err := ReserveHistorySequences(txCtx, tx, []string{"field-z"}); err != nil {
					return err
				}
				close(retryFirstHistoryLocked)
				select {
				case <-allowOutOfOrder:
				case <-ctx.Done():
					return ctx.Err()
				}
				_, err := ReserveHistorySequences(txCtx, tx, []string{"field-a"})
				return err
			}

			// The retry must reach its business lock before taking the forced
			// history-sequence plan. Old code acquired field-a/field-z before
			// executing this callback, creating a Lead <-> sequence deadlock
			// with the normal transaction below.
			if attempt == 2 {
				close(retrySecondStarted)
			}
			var leadID string
			if err := tx.QueryRowContext(txCtx, "SELECT Id FROM `Lead` WHERE Id = ? FOR UPDATE", "lead-001").Scan(&leadID); err != nil {
				return err
			}
			if _, err := ReserveHistorySequences(txCtx, tx, []string{"field-z"}); err != nil {
				return err
			}
			_, err := ReserveHistorySequences(txCtx, tx, []string{"field-a"})
			return err
		}, 3)
	}()

	await(retryFirstHistoryLocked, "retry first history reservation")
	go func() {
		normalErr <- tm.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
			normalAttempts.Add(1)
			tx := tm.ExtractTx(txCtx).(Executor)
			var leadID string
			if err := tx.QueryRowContext(txCtx, "SELECT Id FROM `Lead` WHERE Id = ? FOR UPDATE", "lead-001").Scan(&leadID); err != nil {
				return err
			}
			if normalAttempts.Load() == 1 {
				close(normalLeadLocked)
			}
			select {
			case <-retrySecondStarted:
			case <-ctx.Done():
				return ctx.Err()
			}
			_, err := ReserveHistorySequences(txCtx, tx, []string{"field-a"})
			return err
		}, 3)
	}()

	await(normalLeadLocked, "normal Lead lock")
	close(allowOutOfOrder)

	for name, result := range map[string]<-chan error{
		"normal": normalErr,
		"retry":  retryErr,
	} {
		select {
		case err := <-result:
			if err != nil {
				t.Fatalf("%s transaction: %v", name, err)
			}
		case <-ctx.Done():
			t.Fatalf("timed out waiting for %s transaction: %v", name, ctx.Err())
		}
	}

	if got := normalAttempts.Load(); got != 1 {
		t.Fatalf("normal attempts = %d, want 1; a history-first retry must not force a normal Lead-first write to retry", got)
	}
	if got := retryAttempts.Load(); got != 2 {
		t.Fatalf("retry attempts = %d, want 2; only the planned history re-execution should retry", got)
	}
	for fieldID, want := range map[string]int64{"field-a": 2, "field-z": 1} {
		var current int64
		if err := rawDB.QueryRow(`SELECT CurrentSequence FROM _System_FieldHistorySequence WHERE FieldId = ?`, fieldID).Scan(&current); err != nil {
			t.Fatalf("read %s sequence: %v", fieldID, err)
		}
		if current != want {
			t.Fatalf("%s sequence = %d, want %d", fieldID, current, want)
		}
	}
}