fix(message): retry message writes on MySQL deadlock (1213/1205)

[BenderRodrigez]

Problem

message.Create and CreatePair allocate the per-session sequence number inside their transaction as SELECT COALESCE(MAX(seq),0)+1 WHERE session_id=? → INSERT. When two writes to the same session run concurrently — e.g. a foreground tool result and a background task's synthetic tool result landing together — both take shared range locks on the session_id index, then each needs the insert-intention lock. InnoDB breaks the cycle as a deadlock (Error 1213). There is no retry anywhere in the DB layer, so this transient, self-healing conflict propagates all the way up and kills the whole agent flow at internal/llm/agent/agent.go:

failed to process events: failed to create cancelled tool message:
Error 1213 (40001): Deadlock found when trying to get lock; try restarting transaction

Seen on a real developer-react-on-jira run (MICRO-1014) where the agent had a backgrounded shell task whose synthetic tool-result write raced the foreground turn's tool-result write on one session. SQLite deployments don't hit this (single writer), but the MySQL session provider (SESSION_PROVIDER_TYPE=mysql, used in the pod) does.

Fix

Add db.WithTxRetry — a bounded, ctx-aware retry with full-jitter exponential backoff (5 attempts, 2ms→100ms) that re-runs the transaction on retryable conflicts, matching MySQL's own "try restarting transaction" guidance:

• IsRetryableTxError classifies via *mysql.MySQLError → 1213 (deadlock) and 1205 (lock-wait timeout). It errors.As-unwraps, and returns false for every non-MySQL error, including all SQLite errors, so the SQLite path is untouched.
• The transactional bodies of Create and CreatePair are wrapped in it. Each attempt opens its own tx and defer tx.Rollback()s, so re-running is safe; publish/fromDBItem happen once, after the tx succeeds.

sethvargo/go-retry is only an indirect dep and isn't used anywhere in-tree, so this follows the existing bespoke-backoff style (cf. internal/llm/provider/anthropic.go).

Follow-up (not in this PR)

The durable fix is to remove the SELECT-MAX-then-INSERT race entirely — serialize seq allocation per session (in-process mutex keyed by sessionID) or make it atomic. Retry is the minimal, low-risk fix for the fatal error; happy to do the redesign as a separate change if you'd prefer.

Tests

internal/db/retry_test.go (white-box, no DB needed): error classifier table + retry loop (first-try success, retry-then-succeed, non-retryable passthrough, exhaustion returns last error, already-cancelled ctx makes zero calls). go build ./..., go vet, and go test ./internal/db/... ./internal/message/... all pass locally on go 1.25.8.

Context: this is the fatal half of the MICRO-1014 flow failure. The other half — the agent runtime image lacking a C toolchain so pipenv lock couldn't build uvloop/httptools — is a separate c2-agent/build/agent.dockerfile change.

🤖 Generated with Claude Code

[obukhovaa]

Review summary (verified by live reproduction) — merging as-is

Verdict: bug confirmed, fix is correct and safe — including multiple opencode processes sharing one MySQL DB. Reproduced against mysql:8.0.39 (the same image the c2-agent chart deploys) by replaying message.Create's exact tx body (BEGIN → SELECT MAX(seq) → INSERT → COMMIT) from concurrent same-session writers split across two connection pools — indistinguishable from two processes at the InnoDB lock level:

Scenario	Result
No retry (pre-PR), 8 sustained writers	125/320 inserts failed with 1213, writes lost
WithTxRetry, 2 writers	0 errors, 0 lost writes
WithTxRetry, 4 writers	0 errors, 0 lost writes
WithTxRetry, 8 sustained writers	1–5/320 exhausted all 5 attempts

Root-cause correction (for the record)

The deadlock is not range locks from SELECT MAX(seq) — under REPEATABLE READ that's a non-locking consistent read. The actual mechanism is the update_session_message_count_on_insert AFTER INSERT trigger (initial migration; re-asserted by 20260430120000_repair_message_triggers.sql): every message INSERT takes an S lock on the parent sessions row (FK check), then the trigger needs an X lock on the same row. Two concurrent inserts → both hold S, both wait for X → S→X conversion deadlock. It reproduces even between two autocommit single INSERTs, so in-process seq serialization alone would not eliminate it — the retry is the right primary defense and should stay even after any seq redesign.

Multi-process safety (checked)

• No double-insert risk: on stock MySQL 8, 1213/1205 surface during statement execution, never on a successful commit (1213 rolls back the whole tx; 1205 + defer tx.Rollback() cleans up). Fresh UUID per attempt; pubsub publishes once, after success.
• Seq recomputed per attempt in a fresh snapshot — a retry that lost the race allocates after the winner's committed rows.
• Retries are process-local, bounded, full-jittered; no cross-process coordination needed.
• SQLite path untouched (errors.As on *mysql.MySQLError only).
• Duplicate seq values under concurrent writers remain possible (measured 160 rows → 70 distinct seqs at 4 writers) — pre-existing MAX+1 race, not widened by this PR.

Mechanical checks on the branch: go build ./..., go vet, full go test ./..., new unit tests under -race — all pass.

---

Follow-up TODO (in priority order)

• Durable fix: remove the S→X trigger deadlock at the source. Maintain sessions.message_count in application code inside the same tx, or drop the counter in favor of COUNT(*) (index idx_messages_session_id exists). Note: the PR's proposed "serialize seq allocation in-process" follow-up does not fix this — the deadlock is trigger-based, reproduces between autocommit INSERTs, and crosses processes.
• Wrap the params.Seq != 0 fast path in Create (internal/message/message.go:122-142) with db.WithTxRetry — same trigger fires there (caller: internal/flow/service.go:1214 message copy); an unretried 1213 on that path is still fatal.
• Consider a larger retry budget — at 8 sustained same-session writers, 1–5/320 writes exhausted 5 attempts (~15 ms total expected backoff). Realistic bursts (2–4 writers) showed zero failures, so not urgent; txRetryMaxAttempts=8 and/or base 5–10 ms is near-free insurance given the failure mode kills the whole flow.
• Other unprotected writers on the same sessions row can be picked as the 1213 victim instead and have no retry: session cost updates (UPDATE sessions), message Delete (AFTER DELETE trigger), history service tx (internal/history/file.go:103). Expect occasional 1213s from these until the trigger fix lands.
• Optional: land the reproduction below as a mysql_integration test (make test-mysql) so the deadlock has a permanent regression guard.
• Pre-existing, fold into the seq redesign if desired: duplicate seq under concurrent writers (no unique index on (session_id, seq); ORDER BY seq, created_at tiebreak has second resolution).

Reproduction test used for this review (drop into internal/db/deadlock_repro_test.go, run via make test-mysql or go test -tags=mysql_integration -run TestMySQLConcurrentMessageInsertDeadlock ./internal/db/)

//go:build mysql_integration

// Reproduction for PR #19: proves that concurrent message inserts to the same
// session deadlock on MySQL (pre-retry behavior = fatal error, lost write) and
// that db.WithTxRetry absorbs the conflict with zero lost writes.
//
// The second sql.DB pool models a second opencode process sharing the DB:
// InnoDB locking is per-connection, so two pools are indistinguishable from
// two processes.
package db_test

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

	"github.com/google/uuid"

	"github.com/opencode-ai/opencode/internal/db"
	mysqldb "github.com/opencode-ai/opencode/internal/db/mysql"
)

// insertMessageTx mirrors message.Service.Create's transactional body exactly:
// BEGIN → SELECT COALESCE(MAX(seq),0) for the session → INSERT INTO messages
// (which fires the AFTER INSERT trigger updating sessions.message_count) →
// COMMIT.
func insertMessageTx(ctx context.Context, pool *sql.DB, sessionID string) error {
	tx, err := pool.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()

	var maxSeq int64
	if err := tx.QueryRowContext(ctx,
		"SELECT CAST(COALESCE(MAX(seq), 0) AS SIGNED) FROM messages WHERE session_id = ?",
		sessionID).Scan(&maxSeq); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx,
		`INSERT INTO messages (id, session_id, role, parts, model, seq, synthetic, created_at, updated_at)
		 VALUES (?, ?, 'tool', '[]', NULL, ?, 0, UNIX_TIMESTAMP(), UNIX_TIMESTAMP())`,
		uuid.New().String(), sessionID, maxSeq+1); err != nil {
		return err
	}
	return tx.Commit()
}

func TestMySQLConcurrentMessageInsertDeadlock(t *testing.T) {
	conn := openTestMySQL(t)

	pool2, err := sql.Open("mysql", os.Getenv("TEST_MYSQL_DSN"))
	if err != nil {
		t.Fatalf("open second pool: %v", err)
	}
	t.Cleanup(func() { _ = pool2.Close() })

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	defer cancel()

	q := mysqldb.New(conn)
	for _, id := range []string{"S-noretry", "S-retry-2", "S-retry-4", "S-retry-8"} {
		if _, err := q.CreateSession(ctx, mysqldb.CreateSessionParams{
			ID:        id,
			ProjectID: sql.NullString{String: "proj", Valid: true},
			Title:     "deadlock repro",
		}); err != nil {
			t.Fatalf("CreateSession %s: %v", id, err)
		}
	}

	run := func(sessionID string, workers, perWorker int, withRetry bool) (deadlocks, otherErrs int64) {
		var wg sync.WaitGroup
		var dl, oe int64
		for w := 0; w < workers; w++ {
			pool := conn
			if w%2 == 1 {
				pool = pool2 // odd workers act as the "second process"
			}
			wg.Add(1)
			go func(pool *sql.DB) {
				defer wg.Done()
				for i := 0; i < perWorker; i++ {
					var err error
					if withRetry {
						err = db.WithTxRetry(ctx, func() error {
							return insertMessageTx(ctx, pool, sessionID)
						})
					} else {
						err = insertMessageTx(ctx, pool, sessionID)
					}
					if err != nil {
						if db.IsRetryableTxError(err) {
							atomic.AddInt64(&dl, 1)
						} else {
							atomic.AddInt64(&oe, 1)
							t.Logf("non-retryable error: %v", err)
						}
					}
				}
			}(pool)
		}
		wg.Wait()
		return dl, oe
	}

	stats := func(sessionID string) (rows, distinctSeq int64) {
		if err := conn.QueryRowContext(ctx, "SELECT COUNT(*), COUNT(DISTINCT seq) FROM messages WHERE session_id = ?", sessionID).Scan(&rows, &distinctSeq); err != nil {
			t.Fatalf("stats %s: %v", sessionID, err)
		}
		return rows, distinctSeq
	}

	// Phase 1 — pre-retry behavior: no retry, sustained 8-way contention.
	// Deadlock victims surface as fatal errors and their writes are lost.
	dl, oe := run("S-noretry", 8, 40, false)
	rows1, dseq1 := stats("S-noretry")
	t.Logf("phase1 (no retry, w=8): deadlocks=%d otherErrs=%d rows=%d/320 distinctSeq=%d", dl, oe, rows1, dseq1)
	if dl == 0 {
		t.Errorf("expected at least one deadlock without retry — bug not reproduced")
	}

	// Phase 2 — retry behavior at increasing sustained contention. w=2..4
	// models the realistic case (foreground turn racing background task
	// completions); w=8 is the stress bound, reported but not asserted.
	for _, w := range []int{2, 4, 8} {
		sid := map[int]string{2: "S-retry-2", 4: "S-retry-4", 8: "S-retry-8"}[w]
		total := int64(w * 40)
		dl2, oe2 := run(sid, w, 40, true)
		rows2, dseq2 := stats(sid)
		t.Logf("phase2 (retry, w=%d): exhaustedRetries=%d otherErrs=%d rows=%d/%d distinctSeq=%d", w, dl2, oe2, rows2, total, dseq2)
		if w <= 4 && (dl2 != 0 || oe2 != 0 || rows2 != total) {
			t.Errorf("w=%d with retry: want zero errors and %d rows, got exhausted=%d others=%d rows=%d", w, total, dl2, oe2, rows2)
		}
	}
}

Observed output (mysql:8.0.39, tmpfs, Apple silicon host):

phase1 (no retry, w=8): deadlocks=125 otherErrs=0 rows=195/320 distinctSeq=64
phase2 (retry, w=2): exhaustedRetries=0 otherErrs=0 rows=80/80 distinctSeq=44
phase2 (retry, w=4): exhaustedRetries=0 otherErrs=0 rows=160/160 distinctSeq=70
phase2 (retry, w=8): exhaustedRetries=1 otherErrs=0 rows=319/320 distinctSeq=107

🤖 Reviewed with Claude Code