use std::fs;
use std::path::Path;
#[derive(Debug, Clone)]
pub struct Migration {
pub version: i64,
pub name: String,
pub up: String,
}
pub fn discover(dir: &Path) -> anyhow::Result<Vec<Migration>> {
let mut migrations = Vec::new();
for entry in fs::read_dir(dir)? {
let path = entry?.path();
if path.extension().and_then(|e| e.to_str()) != Some("sql") {
continue;
}
let stem = path
.file_stem()
.and_then(|s| s.to_str())
.ok_or_else(|| anyhow::anyhow!("invalid filename"))?;
let (version_str, name) = stem
.split_once('_')
.ok_or_else(|| anyhow::anyhow!("expected <version>_<name>.sql, got {stem}"))?;
let version: i64 = version_str
.parse()
.map_err(|_| anyhow::anyhow!("non-numeric version in {stem}"))?;
let up = fs::read_to_string(&path)?;
migrations.push(Migration { version, name: name.to_string(), up });
}
migrations.sort_by_key(|m| m.version);
Ok(migrations)
}
use std::collections::HashSet;
use tokio_postgres::Client;
use crate::migration::Migration;
pub async fn ensure_ledger(client: &Client) -> anyhow::Result<()> {
client
.batch_execute(
"CREATE TABLE IF NOT EXISTS schema_migrations (\n version BIGINT PRIMARY KEY,\n name TEXT NOT NULL,\n applied_at TIMESTAMPTZ NOT NULL DEFAULT now()\n )",
)
.await?;
Ok(())
}
async fn applied_versions(client: &Client) -> anyhow::Result<HashSet<i64>> {
let rows = client
.query("SELECT version FROM schema_migrations", &[])
.await?;
Ok(rows.iter().map(|r| r.get::<_, i64>(0)).collect())
}
pub async fn run_pending(client: &mut Client, migrations: &[Migration]) -> anyhow::Result<usize> {
ensure_ledger(client).await?;
let applied = applied_versions(client).await?;
let mut count = 0;
for m in migrations.iter().filter(|m| !applied.contains(&m.version)) {
let tx = client.transaction().await?;
tx.batch_execute(&m.up).await?;
tx.execute(
"INSERT INTO schema_migrations (version, name) VALUES ($1, $2)",
&[&m.version, &m.name],
)
.await?;
tx.commit().await?;
println!("applied {:03}_{}", m.version, m.name);
count += 1;
}
Ok(count)
}
mod migration;
mod runner;
use std::path::Path;
use tokio_postgres::NoTls;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let url = std::env::var("DATABASE_URL")?;
let (mut client, connection) = tokio_postgres::connect(&url, NoTls).await?;
tokio::spawn(async move {
if let Err(e) = connection.await {
eprintln!("connection error: {e}");
}
});
let migrations = migration::discover(Path::new("db/migrations"))?;
let applied = runner::run_pending(&mut client, &migrations).await?;
if applied == 0 {
println!("schema up to date");
} else {
println!("applied {applied} migration(s)");
}
Ok(())
}
This snippet shows a minimal but real migration runner for Postgres built on tokio-postgres. The core idea is the ledger pattern: a dedicated table records which migration versions have already been applied, and the runner only executes the ones missing from that ledger, always in ascending version order. This makes running migrations idempotent — invoking it repeatedly is safe because already-applied versions are skipped — which is exactly the property needed when several app instances boot at once or a deploy is retried.
In migration.rs, a Migration is just a numeric version, a human name, and the raw up SQL. The discover function walks a directory, parses filenames like 001_create_users.sql into a version and name, and returns the list sorted by version. Sorting up front means the caller never has to reason about filesystem ordering, which is not guaranteed to be numeric. Parsing is deliberately strict: a filename that does not split into version_name.sql is rejected rather than silently ignored, so a typo cannot cause a migration to be skipped.
In runner.rs, ensure_ledger creates the schema_migrations table with IF NOT EXISTS, so the very first run bootstraps its own bookkeeping. applied_versions reads the current set into a HashSet for cheap membership checks. The heart is run_pending: it filters the discovered migrations against that set, then for each missing one opens a transaction, executes the up SQL as a batch, inserts the version row, and commits. Wrapping each migration in its own transaction is the key reliability decision — either the DDL and the ledger insert both land, or neither does, so the runner can never mark a version applied without the schema change actually succeeding. On failure the transaction is dropped and rolls back, and the error propagates so the process exits non-zero.
In main.rs, the pieces are wired together: connect, spawn the connection task that tokio-postgres requires, discover, and run. Note the trade-off — migrations run serially and take no global lock here, so in a multi-instance deploy an advisory lock around run_pending would be a sensible addition to avoid two runners racing on the same pending version.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
type User {
id: ID!
name: String!
email: String!
posts: [Post!]!
createdAt: String!GraphQL API with Spring Boot
class SignupForm
include ActiveModel::Model
include ActiveModel::Attributes
attribute :account_name, :string
attribute :email, :stringShallow Controller, Deep Params: Form Object Pattern
use clap::Parser;
#[derive(Parser, Debug)]
#[command(author, version, about)]
struct Args {
#[arg(short, long)]clap for CLI argument parsing with derive macros
Share this code
Here's the card — post it anywhere.
