queue.rs 40.0%

36 of 90 lines run, 8 of 36 functions · run #515 on main · run missed

//! The worker side of the durable work queue (oxygenhub#276,
//! docs/roles.md): every worker process takes due items, a batch at a
//! time, and does them. Work that fails where it starts (a notification's
//! email, the Git mirror after a push) is queued here instead of being
//! lost, and retried with backoff.

use crate::db::{self, queue::Item};
use crate::Shared;
use std::time::Duration;

/// How often an idle worker looks for due items.
pub const POLL: Duration = Duration::from_secs(2);
/// How long a claim lasts before another worker may take the item.
const LEASE_MS: i64 = 5 * 60 * 1000;
/// Items taken at once.
const BATCH: i64 = 10;

/// An email to send again (`mail`).
pub async fn queue_mail(
    app: &Shared,
    to_name: &str,
    to_addr: &str,
    subject: &str,
    body: &str,
    action: Option<(&str, &str)>,
    unsubscribe: Option<&str>,
) {
    let payload = serde_json::json!({
        "to_name": to_name, "to_addr": to_addr, "subject": subject, "body": body,
        "action_label": action.map(|a| a.0), "action_url": action.map(|a| a.1), "unsubscribe": unsubscribe,
    });
    if let Err(e) = db::queue::enqueue(&app.db, "mail", &payload, "", 30_000).await {
        tracing::warn!("queueing an email to {to_addr}: {e}");
    }
}

/// A repository's Git mirror to refresh again (`mirror.refresh`), once.
pub async fn queue_mirror(app: &Shared, repo_id: i64) {
    let payload = serde_json::json!({ "repo_id": repo_id });
    if let Err(e) = db::queue::enqueue(&app.db, "mirror.refresh", &payload, &repo_id.to_string(), 30_000).await {
        tracing::warn!("queueing a Git mirror refresh: {e}");
    }
}

/// This process's name as a claimant: its host and pid.
fn worker_name() -> String {
    let host = std::fs::read_to_string("/etc/hostname").map(|h| h.trim().to_string()).unwrap_or_default();
    format!("{host}:{}", std::process::id())
}

/// Take and do due items until the process stops.
pub async fn worker(app: Shared) {
    let me = worker_name();
    loop {
        let items = match db::queue::claim(&app.db, &me, BATCH, LEASE_MS).await {
            Ok(items) => items,
            Err(e) => {
                tracing::warn!("work queue: claiming: {e}");
                Vec::new()
            }
        };
        let idle = items.is_empty();
        for item in items {
            run(&app, item).await;
        }
        if idle {
            tokio::time::sleep(POLL).await;
        }
    }
}

/// Do one item and record how it went.
pub async fn run(app: &Shared, item: Item) {
    let outcome = match item.kind.as_str() {
        "mail" => mail(app, &item).await,
        "mirror.refresh" => mirror(app, &item).await,
        other => Err(format!("no worker knows {other:?}")),
    };
    let recorded = match outcome {
        Ok(()) => db::queue::done(&app.db, item.id).await,
        Err(e) => {
            tracing::warn!(kind = %item.kind, attempt = item.attempts, "work queue: {e}");
            // An item nobody can do is dead at once.
            let item = if item.kind != "mail" && item.kind != "mirror.refresh" {
                Item { attempts: db::queue::MAX_ATTEMPTS, ..item }
            } else {
                item
            };
            db::queue::failed(&app.db, &item, &e).await
        }
    };
    if let Err(e) = recorded {
        tracing::warn!("work queue: recording an outcome: {e}");
    }
}

async fn mail(app: &Shared, item: &Item) -> Result<(), String> {
    let p: serde_json::Value = serde_json::from_str(&item.payload).map_err(|e| e.to_string())?;
    let s = |k: &str| p[k].as_str().unwrap_or_default().to_string();
    let (label, url) = (s("action_label"), s("action_url"));
    let action = (!url.is_empty()).then_some(crate::mail::Action { label: &label, url: &url });
    let unsubscribe = s("unsubscribe");
    crate::mail::send_letter(
        app,
        &s("to_name"),
        &s("to_addr"),
        &s("subject"),
        &s("body"),
        action,
        (!unsubscribe.is_empty()).then_some(unsubscribe.as_str()),
    )
    .await
}

async fn mirror(app: &Shared, item: &Item) -> Result<(), String> {
    let p: serde_json::Value = serde_json::from_str(&item.payload).map_err(|e| e.to_string())?;
    let id = p["repo_id"].as_i64().ok_or("no repo_id")?;
    // Gone, or its mirror turned off meanwhile: nothing to do.
    let Some(row) = db::repos::get_by_id(&app.db, id).await.map_err(|e| e.to_string())? else { return Ok(()) };
    crate::mirror::refresh(app, &row).await
}

1		//! The worker side of the durable work queue (oxygenhub#276,
2		//! docs/roles.md): every worker process takes due items, a batch at a
3		//! time, and does them. Work that fails where it starts (a notification's
4		//! email, the Git mirror after a push) is queued here instead of being
5		//! lost, and retried with backoff.
6
7		use crate::db::{self, queue::Item};
8		use crate::Shared;
9		use std::time::Duration;
10
11		/// How often an idle worker looks for due items.
12		pub const POLL: Duration = Duration::from_secs(2);
13		/// How long a claim lasts before another worker may take the item.
14		const LEASE_MS: i64 = 5 * 60 * 1000;
15		/// Items taken at once.
16		const BATCH: i64 = 10;
17
18		/// An email to send again (`mail`).
19	0×	pub async fn queue_mail(
20	0×	app: &Shared,
21	0×	to_name: &str,
22	0×	to_addr: &str,
23	0×	subject: &str,
24	0×	body: &str,
25	0×	action: Option<(&str, &str)>,
26	0×	unsubscribe: Option<&str>,
27	0×	) {
28	0×	let payload = serde_json::json!({
29	0×	"to_name": to_name, "to_addr": to_addr, "subject": subject, "body": body,
30	0×	"action_label": action.map(\|a\| a.0), "action_url": action.map(\|a\| a.1), "unsubscribe": unsubscribe,
31		});
32	0×	if let Err(e) = db::queue::enqueue(&app.db, "mail", &payload, "", 30_000).await {
33	0×	tracing::warn!("queueing an email to {to_addr}: {e}");
34	0×	}
35	0×	}
36
37		/// A repository's Git mirror to refresh again (`mirror.refresh`), once.
38	0×	pub async fn queue_mirror(app: &Shared, repo_id: i64) {
39	0×	let payload = serde_json::json!({ "repo_id": repo_id });
40	0×	if let Err(e) = db::queue::enqueue(&app.db, "mirror.refresh", &payload, &repo_id.to_string(), 30_000).await {
41	0×	tracing::warn!("queueing a Git mirror refresh: {e}");
42	0×	}
43	0×	}
44
45		/// This process's name as a claimant: its host and pid.
46	197×	fn worker_name() -> String {
47	197×	let host = std::fs::read_to_string("/etc/hostname").map(\|h\| h.trim().to_string()).unwrap_or_default();
48	197×	format!("{host}:{}", std::process::id())
49	197×	}
50
51		/// Take and do due items until the process stops.
52	197×	pub async fn worker(app: Shared) {
53	197×	let me = worker_name();
54		loop {
55	289×	let items = match db::queue::claim(&app.db, &me, BATCH, LEASE_MS).await {
56	289×	Ok(items) => items,
57	0×	Err(e) => {
58	0×	tracing::warn!("work queue: claiming: {e}");
59	0×	Vec::new()
60		}
61		};
62	289×	let idle = items.is_empty();
63	289×	for item in items {
64	0×	run(&app, item).await;
65		}
66	289×	if idle {
67	289×	tokio::time::sleep(POLL).await;
68	0×	}
69		}
70		}
71
72		/// Do one item and record how it went.
73	2×	pub async fn run(app: &Shared, item: Item) {
74	2×	let outcome = match item.kind.as_str() {
75	2×	"mail" => mail(app, &item).await,
76	2×	"mirror.refresh" => mirror(app, &item).await,
77	1×	other => Err(format!("no worker knows {other:?}")),
78		};
79	2×	let recorded = match outcome {
80	1×	Ok(()) => db::queue::done(&app.db, item.id).await,
81	1×	Err(e) => {
82	1×	tracing::warn!(kind = %item.kind, attempt = item.attempts, "work queue: {e}");
83		// An item nobody can do is dead at once.
84	1×	let item = if item.kind != "mail" && item.kind != "mirror.refresh" {
85	1×	Item { attempts: db::queue::MAX_ATTEMPTS, ..item }
86		} else {
87	0×	item
88		};
89	1×	db::queue::failed(&app.db, &item, &e).await
90		}
91		};
92	2×	if let Err(e) = recorded {
93	0×	tracing::warn!("work queue: recording an outcome: {e}");
94	2×	}
95	2×	}
96
97	0×	async fn mail(app: &Shared, item: &Item) -> Result<(), String> {
98	0×	let p: serde_json::Value = serde_json::from_str(&item.payload).map_err(\|e\| e.to_string())?;
99	0×	let s = \|k: &str\| p[k].as_str().unwrap_or_default().to_string();
100	0×	let (label, url) = (s("action_label"), s("action_url"));
101	0×	let action = (!url.is_empty()).then_some(crate::mail::Action { label: &label, url: &url });
102	0×	let unsubscribe = s("unsubscribe");
103	0×	crate::mail::send_letter(
104	0×	app,
105	0×	&s("to_name"),
106	0×	&s("to_addr"),
107	0×	&s("subject"),
108	0×	&s("body"),
109	0×	action,
110	0×	(!unsubscribe.is_empty()).then_some(unsubscribe.as_str()),
111	0×	)
112	0×	.await
113	0×	}
114
115	1×	async fn mirror(app: &Shared, item: &Item) -> Result<(), String> {
116	1×	let p: serde_json::Value = serde_json::from_str(&item.payload).map_err(\|e\| e.to_string())?;
117	1×	let id = p["repo_id"].as_i64().ok_or("no repo_id")?;
118		// Gone, or its mirror turned off meanwhile: nothing to do.
119	1×	let Some(row) = db::repos::get_by_id(&app.db, id).await.map_err(\|e\| e.to_string())? else { return Ok(()) };
120	0×	crate::mirror::refresh(app, &row).await
121	1×	}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/hub/src/queue.rs 40.0%

Keyboard