Tutoriel backend — « Memo »

Tâches de fond

Pour le travail asynchrone (rappels, exports, nettoyage), un module fait tourner son propre worker. PostgreSQL fournit le verrouillage concurrent sûr via FOR UPDATE SKIP LOCKED.

Le 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(())
}

Le lancer au démarrage #

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

Le worker — dans d'autres langages #

En Rust c'est une tâche tokio ; ailleurs, un démon (systemd) qui boucle. La requête SKIP LOCKED reste identique :

// 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 permet à plusieurs workers de piocher des jobs sans se marcher dessus. Le core dispose aussi d'une file core.jobs ; un module peut soit la réutiliser, soit gérer la sienne comme ici.