SQLx Migrations, Transactions, and Checked Queries

Returning 202 Accepted becomes trustworthy only when ownership survives process death. For a durable run service, the database transaction—not an in-memory channel send—is usually the acceptance boundary.

The companion mosaic-db crate embeds a migration, starts an isolated SQLite database, and commits the run projection with its first ordered event in one transaction. Further events advance both the projection and journal atomically. Six tests cover migration, rollback, concurrency, orphan prevention, domain rejection, and terminal projection updates.

Learning objectives

• design a schema around invariants and access patterns;
• run immutable, versioned migrations safely;
• use SQLx pools, parameter binding, row mapping, and transactions;
• make a projection and event append atomic;
• understand commit, rollback-on-drop, and ambiguous commit outcomes;
• choose runtime queries versus query!/query_as! macros;
• prepare checked-query metadata for hermetic CI;
• test concurrency and rollback against a real database engine;
• distinguish SQLite test value from PostgreSQL production semantics.

Dependencies and mental model

Depends on domain types, Serde, async resource bounds, Axum admission, and resumable events.

validated command
      │
      ▼
BEGIN TRANSACTION
      ├── insert/update run projection
      ├── append ordered event
      ├── enqueue outbox/job (next chapter)
      ▼
COMMIT
      │
      ├── success: service owns the operation
      ├── known failure: no acknowledgement
      └── unknown outcome: reconcile by idempotency key

A transaction gives atomicity inside its database scope. It does not atomically include an HTTP response, provider call, object-store request, process memory, or message broker unless an explicit protocol bridges them.

Schema starts with invariants and reads

The companion migration defines:

CREATE TABLE runs (
    run_id TEXT PRIMARY KEY NOT NULL,
    task_id TEXT NOT NULL,
    objective TEXT NOT NULL,
    input_json TEXT NOT NULL,
    status TEXT NOT NULL CHECK (
        status IN ('queued', 'running', 'completed', 'failed', 'cancelled')
    ),
    last_sequence INTEGER NOT NULL CHECK (last_sequence >= 0),
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE run_events (
    run_id TEXT NOT NULL,
    sequence INTEGER NOT NULL CHECK (sequence > 0),
    kind TEXT NOT NULL,
    data_json TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (run_id, sequence),
    FOREIGN KEY (run_id) REFERENCES runs(run_id) ON DELETE CASCADE
);

CREATE INDEX run_events_resume
    ON run_events (run_id, sequence);

The primary key on (run_id, sequence) encodes ordered uniqueness. The foreign key rejects orphan events. The resume index supports:

WHERE run_id = ? AND sequence > ?
ORDER BY sequence ASC

Schema constraints are defense in depth. Domain constructors provide better errors and prevent invalid in-memory state; database constraints protect against bugs, old binaries, maintenance scripts, and concurrent writers.

Production additions include tenant ownership on every row/key, idempotency keys, schema/event versions, program/model/artifact identities, timestamps with explicit type/time zone, retention class, attempt/lease fields, encrypted references, and indexes derived from measured queries.

Migrations are an append-only deployment protocol

The companion embeds:

sqlx::migrate!("./migrations").run(&pool).await?;

and build.rs emits:

println!("cargo:rerun-if-changed=migrations");

so stable Cargo rebuilds when a new migration appears. .gitattributes forces SQL files to LF because SQLx migration hashes are byte-sensitive across platforms.

Rules:

• never edit a migration that may have run in a shared environment;
• add a new migration;
• give application versions a compatibility window;
• make large data backfills separate, resumable, observable jobs;
• keep DDL lock/runtime impact understood;
• gate startup/readiness while required migrations are missing;
• allow one controlled migrator, not every replica racing blindly;
• backup and rehearse rollback/forward recovery.

Expand and contract

For a breaking field change:

1. expand: add nullable/new column/table/index;
2. deploy code that can read old and new, writes both where needed;
3. backfill in bounded resumable batches;
4. verify completeness and query plans;
5. switch reads;
6. contract: later remove old field/compatibility.

A database rollback does not automatically make an old binary compatible with new rows/events. Rollout and schema compatibility are one design.

Pool is a capacity limiter

SQLx Pool is cloned cheaply and manages connections. Size it from database capacity and replica count:

maximum database connections
  ≥ replicas × pool maximum
  + migrations/admin/worker reserve

A 50-connection pool across 40 replicas asks for 2,000 connections. Pool wait is queueing and needs a deadline/metric. Long transactions hold connections, versions/locks, and sometimes memory.

The companion in-memory repository deliberately uses one SQLite connection:

SqlitePoolOptions::new()
    .max_connections(1)
    .connect_with(options)
    .await?;

Separate sqlite::memory: connections are separate databases. This is a test topology choice, not a recommended production pool size.

Parameterized queries

The companion uses:

sqlx::query(
    "INSERT INTO runs \
     (run_id, task_id, objective, input_json, status, last_sequence) \
     VALUES (?, ?, ?, ?, 'queued', 1) \
     ON CONFLICT(run_id) DO NOTHING",
)
.bind(run_id.as_str())
.bind(task.id.as_str())
.bind(&task.objective)
.bind(input_json)

Bind values; never concatenate untrusted input into SQL. Identifiers/order clauses cannot normally be bind parameters. Choose them from a closed enum/allowlist or build separate static queries.

Runtime query/query_as still use prepared parameterized SQL and typed decode at runtime. #[derive(sqlx::FromRow)] maps stable columns to StoredRun. Select explicit columns, not SELECT *, so schema additions do not silently change transport or decode assumptions.

Atomic run admission

RunRepository::create_run:

1. validates the Task;
2. serializes the already-validated input;
3. begins a transaction;
4. inserts the run with ON CONFLICT DO NOTHING;
5. checks exactly one row was inserted;
6. inserts sequence-one run_queued;
7. commits;
8. returns the stored projection.

let mut transaction = self.pool.begin().await?;

let inserted = sqlx::query("INSERT ... ON CONFLICT DO NOTHING")
    .bind(run_id.as_str())
    .execute(&mut *transaction)
    .await?;

if inserted.rows_affected() != 1 {
    transaction.rollback().await?;
    return Err(DbError::DuplicateRun);
}

sqlx::query("INSERT INTO run_events ...")
    .bind(run_id.as_str())
    .execute(&mut *transaction)
    .await?;

transaction.commit().await?;

If the second insert fails, dropping/rolling back the transaction removes the first. The duplicate test proves existing task/event data is not overwritten or appended.

In the next chapter the transaction also creates an idempotency record and durable job/outbox row. Only then should the Axum handler return 202.

Atomic event and projection update

The canonical journal answers “what happened”; the projection answers “what is current?” Both must move together:

let sequence = sqlx::query_scalar::<_, i64>(
    "UPDATE runs \
     SET last_sequence = last_sequence + 1 \
     WHERE run_id = ? \
     RETURNING last_sequence",
)
.bind(run_id.as_str())
.fetch_optional(&mut *transaction)
.await?
.ok_or(DbError::RunNotFound)?;

// update projection status when this event changes it
// insert event at the returned sequence
transaction.commit().await?;

The update allocates a unique sequence under database serialization. Eight concurrent companion tasks append progress and the resulting event IDs are exactly 1..=9.

Production transition SQL should additionally require expected state/version/lease owner:

UPDATE runs
SET status = 'completed',
    version = version + 1,
    last_sequence = last_sequence + 1
WHERE run_id = $1
  AND status = 'running'
  AND version = $2
  AND lease_owner = $3
RETURNING version, last_sequence;

Zero rows means conflict/lost lease/invalid state, not success.

Commit outcomes and idempotency

Three outcomes matter:

1. commit returned success;
2. commit returned a known failure before commit;
3. connection disappeared around commit and outcome is unknown.

The third case is why “retry the INSERT” needs idempotency. Reconnect and query by a stable tenant-scoped idempotency key/request fingerprint. Do not generate a new run and charge twice.

Transactions cannot make a provider call atomic. Use:

• transactional outbox/job row committed with run;
• worker claims later;
• idempotent provider/tool operations where available;
• effect ledger/reconciliation for ambiguous external outcomes.

Transaction length and isolation

Keep transactions around database work only:

bad:
BEGIN → select → call model for 45 s → update → COMMIT

good:
BEGIN → claim/record attempt → COMMIT
call model outside transaction
BEGIN → verify lease/version → record result/event → COMMIT

Do not wait on providers, GPU, uploads, or users while holding a transaction.

Isolation choice depends on invariants:

• read committed may allow changing repeated reads;
• repeatable read provides a stable snapshot but can still require conflict handling;
• serializable prevents more anomalies but may abort/retry;
• row/advisory locks coordinate specific resources;
• unique/check/foreign constraints are often the strongest simple defense.

Every retry loop needs a maximum, jitter/backoff where relevant, an idempotent transaction body, and deadline.

Runtime queries versus compile-time macros

SQLx offers two styles.

Runtime query builders

sqlx::query_as::<_, StoredRun>("SELECT ... WHERE run_id = ?")

Advantages:

• ordinary compilation needs no development database/query cache;
• named FromRow type;
• dynamic optional clauses are possible;
• simple for the self-contained SQLite companion.

Costs:

• SQL syntax/schema/column compatibility checked when tests/run;
• nullable/type mismatch can remain runtime failure.

query! and query_as!

Macros ask the actual database to analyze literal SQL at build/prepare time and check bind/output shape. They are database-specific and powerful, but build reproducibility needs a deliberate workflow.

Online:

DATABASE_URL=postgres://... cargo check

Offline:

cargo sqlx prepare --workspace -- --all-targets --all-features
git add .sqlx
SQLX_OFFLINE=true cargo check --workspace --all-targets --all-features
cargo sqlx prepare --check --workspace -- --all-targets --all-features

Commit .sqlx metadata. CI checks it is current. Do not let an accidental production DATABASE_URL become a build dependency. Query macro analysis reflects the schema used during prepare; migrations and metadata must move together.

The companion intentionally uses runtime queries plus real migrated integration tests so readers can clone/build without SQLx CLI or external database. A production PostgreSQL variant should add checked-query metadata for stable static queries.

SQLite versus PostgreSQL tests

SQLite in-memory tests are fast and prove:

• SQL parses under SQLite;
• migrations apply;
• transaction rollback/commit logic;
• constraints/index shape defined for SQLite;
• repository mapping and cursor behavior;
• application concurrency through the configured pool.

They do not prove PostgreSQL:

• types/nullability/query plans;
• row-lock/isolation behavior;
• SKIP LOCKED, advisory locks, LISTEN/NOTIFY;
• timestamp/JSONB/array semantics;
• connection/TLS/network failure;
• migration lock duration;
• production concurrency/deadlocks.

If production uses PostgreSQL, run integration/chaos/load gates on a real supported PostgreSQL version. Do not market a SQLite-only test as database portability.

JSON columns and typed schema

JSON is useful for modality/provider-specific event payloads, but keep query-critical invariants in typed columns:

• run/tenant/status/sequence/model/program/artifact IDs;
• cost/usage fields used for quota;
• lease/attempt/idempotency;
• retention/classification;
• created/updated/deadline.

Version JSON envelopes and validate after decode. Avoid frequently querying arbitrary nested JSON without indexes/design. Never store raw provider responses/prompts by default when privacy/retention does not permit it.

Runnable companion implementation

cd rust/rust-ai-engineering
cargo run -q -p mosaic-db --example transactional_run
cargo test -p mosaic-db

Expected:

run-sqlx-demo status=completed last_sequence=3 events=3

The exact code is under crates/mosaic-db; the migration is migrations/0001_runs_and_events.sql.

Failure investigation: projection says completed but terminal event is absent

1. query projection and event journal under the same run/tenant;
2. inspect code path for separate transactions/autocommit;
3. check triggers/constraints and transaction error handling;
4. identify manual/old-writer paths;
5. rebuild projection from journal or write a reconciled repair event under policy;
6. move projection and append into one transaction;
7. add failure injection between statements.

Do not insert a fabricated terminal event without provenance.

Failure investigation: migration succeeded on test but production deploy stalls

1. inspect DDL lock and table/index size;
2. identify long transactions blocking schema change;
3. compare engine/version/settings;
4. stop rollout or use compatible old code;
5. apply expand/backfill/contract with online/concurrent index method where supported;
6. rehearse on production-scale copy and record duration/locks;
7. add migration budget/observability and a deployment runbook.

Failure investigation: compile-time query fails only in CI

1. determine online versus offline macro mode;
2. inspect accidental DATABASE_URL/.env;
3. confirm .sqlx metadata is committed at workspace root;
4. run cargo sqlx prepare --check --workspace with all target/features;
5. confirm metadata was generated against current migrations/database version;
6. avoid platform line-ending migration hash drift;
7. pin SQLx CLI/toolchain versions in CI.

Security and performance implications

Security:

• parameterize values and allowlist dynamic identifiers;
• enforce tenant in primary/unique/foreign keys and every query;
• database role uses least privilege; migrator role is separate;
• TLS/credential rotation and secret redaction apply;
• encrypt/limit sensitive JSON and backups;
• prevent cross-tenant existence leaks;
• migrations and repair scripts are privileged production code.

Performance:

• pool capacity is finite and measured;
• transactions remain short;
• indexes follow access pattern and write amplification is measured;
• paginate/stream bounded result sets; never fetch_all unbounded history;
• batch event reads/appends when semantics allow;
• inspect query plans with production-like data;
• JSON serialization/storage and duplicate projection+journal writes cost space/CPU;
• retention/partition/vacuum policy belongs in capacity planning.

Exercises

1. Add tenant_id so cross-tenant run/event queries are structurally difficult.
2. Add optimistic version and reject a stale terminal update.
3. Inject failure after the run insert and prove no run/event commits.
4. Add a PostgreSQL integration profile and compare concurrent sequence allocation.
5. Convert one static query to query_as!, prepare offline metadata, and add the CI check.
6. Add keyset pagination for events with maximum page size.
7. Write an expand/backfill/contract plan for splitting input_json.

Solution guidance

Use composite identities/foreign keys that include tenant. A stale update uses WHERE version = ? and treats zero rows as conflict. Failure injection should occur inside the transaction before commit; then query with a separate connection/repository. Keyset pagination is sequence > cursor ORDER BY sequence LIMIT bounded_limit; never use offset for a growing journal. Generate checked-query metadata against the exact supported engine/schema and commit it.

Check your understanding

• When is 202 truthful?
• What does a transaction not include?
• Why select explicit columns?
• What happens when an SQLx transaction drops without commit?
• Why is a commit error sometimes ambiguous?
• What do SQLite tests fail to prove about PostgreSQL?
• How does offline query verification remain current?
• Why should long provider work occur outside transactions?

Completion checklist

• Migrations are immutable, versioned, and reproducible.
• Schema constraints encode critical invariants.
• Run projection and first event commit atomically.
• Projection changes and event append commit atomically.
• Dynamic values are bound, identifiers allowlisted.
• Pool/transaction time is bounded and measured.
• Idempotency handles unknown commit outcomes.
• Checked-query online/offline workflow is explicit.
• Production-engine integration tests exist.
• Migration rollout/backfill/rollback is rehearsed.
• I ran the SQLx example and six tests.

Authoritative references

• SQLx documentation
• SQLx transactions
• SQLx query! and offline mode
• SQLx embedded migrations
• SQLx CLI prepare workflow
• SQLite foreign keys
• PostgreSQL transaction isolation
• PostgreSQL explicit locking