Background jobs

For asynchronous work (reminders, exports, cleanup), a module runs its own worker. PostgreSQL provides safe concurrent locking via FOR UPDATE SKIP LOCKED.

The worker #

// services/reminder.rs — chaque module gère son propre travail de fond (polling)
pub async fn run_worker(state: Arc<AppState>) {
    loop {
        tokio::time::sleep(Duration::from_secs(60)).await;
        if let Err(e) = process_due(&state).await {
            tracing::error!(error = %e, "worker memo");
        }
    }
}

async fn process_due(state: &AppState) -> Result<()> {
    let due: Vec<(Uuid,)> = sqlx::query_as(
        "SELECT id FROM memo.jobs WHERE status='pending' AND run_after <= NOW()
         ORDER BY run_after FOR UPDATE SKIP LOCKED LIMIT 50"
    ).fetch_all(&state.db).await?;
    for (id,) in due {
        // … traiter, puis :
        sqlx::query("UPDATE memo.jobs SET status='done', done_at=NOW() WHERE id=$1")
            .bind(id).execute(&state.db).await?;
    }
    Ok(())
}

Starting it at boot #

// dans main.rs, avant axum::serve :
tokio::spawn(reminder::run_worker(Arc::new(state.clone())));

The worker — in other languages #

In Rust it is a tokio task; elsewhere, a looping (systemd) daemon. The SKIP LOCKED query stays the same:

// Rust — tâche tokio
tokio::spawn(async move {
    loop {
        tokio::time::sleep(Duration::from_secs(60)).await;
        let due = sqlx::query("SELECT id FROM memo.jobs
            WHERE status='pending' AND run_after <= NOW()
            FOR UPDATE SKIP LOCKED LIMIT 50").fetch_all(&db).await;
        // … traiter, puis UPDATE … SET status='done'
    }
});

<?php
// worker.php — démon lancé par systemd : php worker.php
while (true) {
    $rows = $pdo->query("SELECT id FROM memo.jobs
        WHERE status='pending' AND run_after <= NOW()
        FOR UPDATE SKIP LOCKED LIMIT 50")->fetchAll();
    foreach ($rows as $r) { /* traiter, puis marquer 'done' */ }
    sleep(60);
}

# Python — boucle de worker
import time
while True:
    rows = db.execute("""SELECT id FROM memo.jobs
        WHERE status='pending' AND run_after <= NOW()
        FOR UPDATE SKIP LOCKED LIMIT 50""").fetchall()
    for r in rows:
        ...  # traiter, puis status='done'
    time.sleep(60)

// Go — goroutine
go func() {
    for {
        time.Sleep(60 * time.Second)
        rows, _ := db.Query(`SELECT id FROM memo.jobs
            WHERE status='pending' AND run_after <= NOW()
            FOR UPDATE SKIP LOCKED LIMIT 50`)
        // … traiter, puis UPDATE … SET status='done'
        rows.Close()
    }
}()

# Perl — boucle de démon
while (1) {
    my $due = $dbh->selectall_arrayref(q{
        SELECT id FROM memo.jobs
        WHERE status='pending' AND run_after <= NOW()
        FOR UPDATE SKIP LOCKED LIMIT 50}, { Slice => {} });
    # ... traiter, puis status='done'
    sleep 60;
}

Note

SKIP LOCKED lets several workers pick jobs without stepping on each other. The core also has a core.jobs queue; a module can either reuse it or manage its own, as here.