//! The group-admin queue, and the drainer that empties it. //! //! //! //! # Why the app has one //! //! `create_group`, `add_member` and `remove_member` each open a conversation //! with a server. A described route handler is synchronous, by //! `quasi_router`'s Decision 6, which exists so egui and a terminal need no //! runtime. So Settings > Sharing could show the groups and could not change //! them. //! //! Queueing is a local write, so it can be described. This is the outbox's //! argument applied a second time, and Max's ruling behind that one ("have //! GoingsOn use an outbox model explicitly", `a3c76a24`) is what makes it a //! pattern rather than a workaround repeated. See [`crate::outbox`], which this //! is deliberately shaped like: a reader who has read that one already knows how //! this starts, stops, backs off and survives a failing tick. //! //! # What a queue buys here, which is not what it bought for mail //! //! There is no send-later value in creating a group. What there is: //! //! - the write survives being offline, instead of failing at the instant //! somebody pressed the button. That matters more for these than for mail, //! because they are rare and deliberate and nobody retries them by habit; //! - a failure is a row with a reason on it, sitting in the section that caused //! it, rather than a toast that has already gone. For `add_member` that is the //! difference between "it did not work" and "it did not work, and here is //! what the server said"; //! - `add_member` needs the master key loaded to seal the group key to the new //! member. A queue turns "you cannot do this right now" into "this happens //! once you unlock", which is honest and is not something a button could //! offer. //! //! # Nothing here is described, and that is the design //! //! The description says "queue this". What drains the queue is not a screen and //! has no address. That division is why a queue answers the async problem rather //! than moving it: the async lives out here, where there has always been a //! runtime. //! //! # What a failed attempt does //! //! Stamps the error, counts the attempt, and leaves the row queued, backing off //! on the count exactly as the outbox does. A row that can never succeed sits //! with its reason on it rather than being deleted, because an intention to add //! somebody to a group that silently disappeared is worse than one still visible //! and failing. use std::sync::Arc; use tauri::Manager; use tokio::time::{Duration, interval}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info}; use crate::state::{AppState, DESKTOP_USER_ID}; /// How often the drainer wakes. /// /// Matches [`crate::outbox`], and for the same reason: it is what makes "queue" /// acceptable as the only way to act. A group is created within a minute of /// being asked for, which is not immediate and is not a wait anybody watches. const CHECK_INTERVAL_SECS: u64 = 60; /// One queued admin write. /// /// [`Default`] rather than eight literal `None`s at every call site: there are /// eight kinds now and no kind reads more than three of the payload columns, so /// a construction that names only what it uses says which those are. #[derive(Debug, Clone, Default)] pub struct QueuedOp { pub id: String, pub kind: String, pub group_id: Option, pub name: Option, pub email: Option, pub pubkey: Option, pub member_user_id: Option, /// The invitation `confirm_invite` and `revoke_invite` address. pub invitation_id: Option, /// A code the user pasted, for `preview_invite` and `accept_invite`. /// /// Never an *issued* token: that one is returned once and is written /// straight into the directory by the drainer. See migration 071. pub invite_token: Option, /// How long an issued invitation should stand, for `create_invite`. pub expires_in_hours: Option, pub attempts: i32, pub last_error: Option, pub done_at: Option, } impl QueuedOp { /// How this reads in the section that queued it. /// /// Written here rather than in the screen because the screen draws a row and /// this is what the row says: a person who queued three of these wants to /// know which is which, and the payload is the only thing that tells them /// apart. #[must_use] pub fn describe(&self) -> String { self.describe_confirming(None) } /// The same, with the fingerprint a queued confirm is authorizing. /// /// A confirm carries a group and an invitation id and deliberately carries /// no fingerprint, because the drainer must re-read the server's answer /// rather than act on a copy (migration 071). But "Confirm an invitation" /// authorizes nothing legible: the fingerprint is the whole content of the /// decision, and a queue row that does not name it asks the reader to take /// the pending act on trust for the minute it sits there. /// /// So the screen resolves it from the directory as it draws and passes it /// in. That is a display of the current answer rather than a second copy of /// it, which is why it lives in the argument and not in the row. #[must_use] pub fn describe_confirming(&self, fingerprint: Option<&str>) -> String { match self.kind.as_str() { "create_group" => format!( "Create the group {}", self.name.as_deref().unwrap_or("(unnamed)") ), "add_member" => format!( "Add {} to a group", self.email.as_deref().unwrap_or("(no address)") ), "remove_member" => "Remove a member from a group".to_owned(), "create_invite" => "Issue an invite code".to_owned(), "revoke_invite" => "Cancel an invitation".to_owned(), "confirm_invite" => fingerprint.map_or_else( // The invitation has left `accepted` since it was queued, or // the directory has not caught up. Both are honest reasons not // to be able to name the fingerprint, and neither is a reason // to name a stale one. || "Admit somebody, once their fingerprint is checked".to_owned(), |fingerprint| format!("Admit the holder of {fingerprint}"), ), "preview_invite" => "Read what an invite code leads to".to_owned(), "accept_invite" => "Accept an invite code".to_owned(), // A kind this build does not know, held rather than refused. It can // only come from a newer build that queued it, and saying so is // better than drawing a blank row. other => format!("An action this version does not understand ({other})"), } } } /// Queue an admin write. The whole of what a described handler does. pub fn enqueue(state: &AppState, op: &QueuedOp) -> Result<(), String> { let conn = state.db.conn().map_err(|e| e.to_string())?; conn.execute( "INSERT INTO group_admin_queue \ (id, user_id, kind, group_id, name, email, pubkey, member_user_id, \ invitation_id, invite_token, expires_in_hours) \ VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", rusqlite::params![ op.id, DESKTOP_USER_ID.to_string(), op.kind, op.group_id, op.name, op.email, op.pubkey, op.member_user_id, op.invitation_id, op.invite_token, op.expires_in_hours, ], ) .map_err(|e| e.to_string())?; Ok(()) } /// Everything still waiting, oldest first, plus anything done that the directory /// has not caught up with yet. /// /// Both, because the moment between "the server accepted it" and "a sync brought /// the group back" is real and a screen that showed neither would look like it /// lost the request. pub fn pending(state: &AppState) -> Result, String> { let conn = state.db.conn().map_err(|e| e.to_string())?; let mut stmt = conn .prepare( "SELECT id, kind, group_id, name, email, pubkey, member_user_id, \ invitation_id, invite_token, expires_in_hours, \ attempts, last_error, done_at \ FROM group_admin_queue WHERE user_id = ?1 ORDER BY queued_at", ) .map_err(|e| e.to_string())?; let rows = stmt .query_map(rusqlite::params![DESKTOP_USER_ID.to_string()], |row| { Ok(QueuedOp { id: row.get(0)?, kind: row.get(1)?, group_id: row.get(2)?, name: row.get(3)?, email: row.get(4)?, pubkey: row.get(5)?, member_user_id: row.get(6)?, invitation_id: row.get(7)?, invite_token: row.get(8)?, expires_in_hours: row.get(9)?, attempts: row.get(10)?, last_error: row.get(11)?, done_at: row.get(12)?, }) }) .map_err(|e| e.to_string())? .collect::, _>>() .map_err(|e| e.to_string())?; Ok(rows) } /// Take a queued write back out. /// /// The way out of a row that will never succeed, and the reason a failure is /// held rather than deleted: the person who queued it decides, not the drainer. pub fn cancel(state: &AppState, id: &str) -> Result { let conn = state.db.conn().map_err(|e| e.to_string())?; let changed = conn .execute( "DELETE FROM group_admin_queue WHERE id = ?1 AND user_id = ?2 AND done_at IS NULL", rusqlite::params![id, DESKTOP_USER_ID.to_string()], ) .map_err(|e| e.to_string())?; Ok(changed > 0) } /// Sweep the rows the server has accepted and the directory has caught up with. /// /// A done row is kept until the group it made is in the directory, so the /// section can say "created" for the moment between the two. Once the directory /// has it, the row has nothing left to say. fn sweep_settled(state: &AppState) { let Ok(conn) = state.db.conn() else { return }; // `create_group` is the only kind whose landing is observable in the // directory. The other two change a member list, which is only fetched for // groups this user administers and may legitimately not have refreshed yet, // so they are swept on age instead. let _ = conn.execute( "DELETE FROM group_admin_queue \ WHERE done_at IS NOT NULL \ AND (kind = 'create_group' \ AND EXISTS (SELECT 1 FROM sync_groups WHERE name = group_admin_queue.name) \ OR done_at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-1 hour'))", [], ); } fn record_failure(state: &AppState, id: &str, message: &str) { let Ok(conn) = state.db.conn() else { return }; if let Err(error) = conn.execute( "UPDATE group_admin_queue SET attempts = attempts + 1, last_error = ?2 WHERE id = ?1", rusqlite::params![id, message], ) { error!("Group queue: could not record the failure: {error}"); } } fn record_done(state: &AppState, id: &str) { let Ok(conn) = state.db.conn() else { return }; if let Err(error) = conn.execute( "UPDATE group_admin_queue SET done_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), \ last_error = NULL WHERE id = ?1", rusqlite::params![id], ) { error!("Group queue: could not record the success: {error}"); } } /// Start the drainer. Runs until cancelled. pub async fn start_group_queue_drainer(app: tauri::AppHandle, cancel: CancellationToken) { let mut check_interval = interval(Duration::from_secs(CHECK_INTERVAL_SECS)); let mut tick: u64 = 0; info!("Group queue drainer started (checking every {CHECK_INTERVAL_SECS} seconds)"); loop { tokio::select! { () = cancel.cancelled() => { info!("Group queue drainer shutting down"); break; } _ = check_interval.tick() => {} } tick = tick.wrapping_add(1); let Some(state) = app.try_state::>() else { debug!("Group queue drainer: state not ready yet"); continue; }; let state: Arc = state.inner().clone(); drain_once(&state, tick).await; } } /// One pass over the queue. /// /// Split from the loop so a test can run a pass without a Tauri handle or a /// minute of waiting, exactly as [`crate::outbox::drain_once`] is. pub async fn drain_once(state: &Arc, tick: u64) { sweep_settled(state); let queued = match pending(state) { Ok(queued) => queued, Err(error) => { error!("Group queue drainer: could not read the queue: {error}"); return; } }; for op in queued { if op.done_at.is_some() { continue; } // The outbox's curve, shared rather than re-derived: a row that has // failed once is retried on the next wake, one that has failed six times // every half hour or so, and the cap stops a long failure from becoming // a silent drop wearing a backoff's clothes. if !crate::outbox::due_on_tick(op.attempts, tick) { continue; } // Not configured is not a failure to count: a device that has never // signed in will never succeed at any of these, and counting attempts // against it would back the row off to never while the reason stays the // same. Left untouched, so it goes as soon as sync is set up. let Some(client) = state.read_recovering() else { continue; }; // Opened per row rather than held across the loop: keeping a pooled // connection in hand across an await takes it out of the pool for the // length of a network call, which is the shape that starves a pool. let mut conn = state.db.conn().ok(); let outcome = perform(&client, &op, conn.as_deref_mut()).await; drop(conn); match outcome { Ok(()) => record_done(state, &op.id), Err(message) => record_failure(state, &op.id, &message), } } } /// Perform one queued write against the server. async fn perform( client: &synckit_client::SyncKitClient, op: &QueuedOp, conn: Option<&mut rusqlite::Connection>, ) -> Result<(), String> { let group = |raw: &Option| -> Result { raw.as_deref() .ok_or_else(|| "No group on the queued action.".to_owned())? .parse::() .map(synckit_client::GroupId::new) .map_err(|_| "The queued action names a group id that will not parse.".to_owned()) }; let invitation = |raw: &Option| -> Result { raw.as_deref() .ok_or_else(|| "No invitation on the queued action.".to_owned())? .parse::() .map(synckit_client::InvitationId::new) .map_err(|_| "The queued action names an invitation id that will not parse.".to_owned()) }; // Already normalised by the handler that queued it, deliberately: there is // one accepted spelling of a pasted code and `normalize_invite_token` owns // it. Re-normalising here would be a second place for the format to live. // // A function rather than a closure beside the two above, because it borrows // out of its argument and a closure cannot state that the returned `&str` // outlives the call. fn token(raw: Option<&str>) -> Result<&str, String> { raw.ok_or_else(|| "No invite code on the queued action.".to_owned()) } match op.kind.as_str() { "create_group" => { let name = op .name .as_deref() .ok_or_else(|| "No name on the queued action.".to_owned())?; let group = client.create_group(name).await.map_err(|e| e.to_string())?; // Into the directory here, the same as `commands::group::group_create` // does on its own path: the group is real on the server, and this // device would otherwise not know its name until the next cycle, so // the section that queued it would show nothing landing. if let Some(conn) = conn { let known = synckit_client::store::directory::KnownGroup { id: group.id, name: group.name, gck_version: group.gck_version, is_admin: true, }; if let Err(error) = synckit_client::store::directory::add_group(conn, &known) { // Not a failure of the write: the group exists. The next // cycle writes the whole directory anyway. error!("Group queue: could not record the new group: {error}"); } } Ok(()) } "add_member" => { let email = op .email .as_deref() .ok_or_else(|| "No address on the queued action.".to_owned())?; let pubkey = op .pubkey .as_deref() .ok_or_else(|| "No public key on the queued action.".to_owned())?; client .add_member(group(&op.group_id)?, email, pubkey) .await .map_err(|e| e.to_string()) } "remove_member" => { let member = op .member_user_id .as_deref() .ok_or_else(|| "No member on the queued action.".to_owned())? .parse::() .map(synckit_client::UserId::new) .map_err(|_| "The queued action names a user id that will not parse.".to_owned())?; client .remove_member(group(&op.group_id)?, member) .await .map_err(|e| e.to_string()) } "create_invite" => { let group = group(&op.group_id)?; let invitation = client .create_invitation(group, op.expires_in_hours) .await .map_err(|e| e.to_string())?; // Not a nicety, unlike `create_group`'s `add_group` above. The // server keeps only a hash of the token, so what came back is the // only copy that will ever exist and this is the one moment it can // be written down. A failure here has issued an invitation whose // code nobody holds, which is why it is reported as a failure of // the write rather than logged past. let conn = conn.ok_or_else(|| { "The invite was issued and could not be written down: no database \ connection. Revoke it from the group and issue another." .to_owned() })?; synckit_client::store::directory::record_issued(conn, group, &invitation).map_err( |error| { format!( "The invite was issued and could not be written down ({error}). \ Revoke it from the group and issue another." ) }, ) } "revoke_invite" => client .revoke_invitation(group(&op.group_id)?, invitation(&op.invitation_id)?) .await .map_err(|e| e.to_string()), // No fingerprint is passed, and none is stored. The queue row names the // invitation and the server answers with the key it currently holds; a // fingerprint copied at queue time would be authorizing a value nobody // re-checked. "confirm_invite" => client .confirm_invitation(group(&op.group_id)?, invitation(&op.invitation_id)?, None) .await .map_err(|e| e.to_string()), "preview_invite" => { let token = token(op.invite_token.as_deref())?; let preview = client .preview_invitation(token) .await .map_err(|e| e.to_string())?; let conn = conn.ok_or_else(|| { "Read the code and could not write down the answer: no database connection." .to_owned() })?; let known = synckit_client::store::directory::KnownPreview { token: token.to_owned(), group_name: preview.group_name, inviter_email: preview.inviter_email, redeemable: preview.redeemable, state: preview.state, expires_at: preview.expires_at.to_rfc3339(), }; synckit_client::store::directory::write_preview(conn, &known) .map_err(|error| format!("Could not write down what the code leads to: {error}")) } "accept_invite" => { client .accept_invitation(token(op.invite_token.as_deref())?) .await .map_err(|e| e.to_string())?; // The code has done its whole job. Clearing it is what takes the // preview section off the screen, so a failure here would leave an // Accept control offering an act that already happened. if let Some(conn) = conn && let Err(error) = synckit_client::store::directory::clear_preview(conn) { error!("Group queue: could not forget the accepted code: {error}"); } Ok(()) } // Held rather than refused, and the message says why so it does not read // as a bug. Only a newer build could have written it. other => Err(format!( "This version does not know how to perform `{other}`. It is kept, not lost." )), } } #[cfg(test)] mod tests { use super::*; async fn state() -> Arc { let (state, _) = crate::test_utils::setup_test_state().await; state } fn op(kind: &str, id: &str) -> QueuedOp { QueuedOp { id: id.to_owned(), kind: kind.to_owned(), group_id: Some("00000000-0000-0000-0000-000000000001".to_owned()), name: Some("The Firm".to_owned()), email: Some("them@localhost".to_owned()), pubkey: Some("k".to_owned()), member_user_id: Some("00000000-0000-0000-0000-000000000002".to_owned()), ..Default::default() } } /// A person who queued three of these wants to know which is which, and the /// payload is the only thing that tells them apart. #[test] fn each_kind_says_what_it_will_do() { assert_eq!( op("create_group", "a").describe(), "Create the group The Firm" ); assert_eq!( op("add_member", "a").describe(), "Add them@localhost to a group" ); assert_eq!( op("remove_member", "a").describe(), "Remove a member from a group" ); } /// Only a newer build could have written it, so saying so beats a blank row /// and beats refusing to draw the queue at all. #[test] fn a_kind_from_a_newer_build_still_reads_as_something() { let described = op("invite_member", "a").describe(); assert!(described.contains("does not understand"), "{described}"); assert!(described.contains("invite_member"), "{described}"); } #[tokio::test] async fn a_queued_action_comes_back_out_in_the_order_it_went_in() { let state = state().await; enqueue(&state, &op("create_group", "first")).unwrap(); enqueue(&state, &op("add_member", "second")).unwrap(); let queued = pending(&state).unwrap(); assert_eq!(queued.len(), 2); assert_eq!(queued[0].id, "first"); assert_eq!(queued[1].id, "second"); } /// A device that has never signed in will never succeed at any of these, and /// counting attempts against it would back the row off to never while the /// reason stays the same. It has to go as soon as sync is set up. #[tokio::test] async fn a_pass_with_no_client_leaves_the_row_untouched() { let state = state().await; enqueue(&state, &op("create_group", "waiting")).unwrap(); drain_once(&state, 1).await; let queued = pending(&state).unwrap(); assert_eq!(queued.len(), 1, "still queued"); assert_eq!(queued[0].attempts, 0, "and not counted against"); assert!(queued[0].last_error.is_none()); } #[tokio::test] async fn cancelling_takes_it_out_and_says_whether_it_did() { let state = state().await; enqueue(&state, &op("create_group", "mistake")).unwrap(); assert!(cancel(&state, "mistake").unwrap()); assert!(pending(&state).unwrap().is_empty()); assert!(!cancel(&state, "mistake").unwrap(), "twice is not a lie"); } /// A cancel that raced the drainer must not report that it undid anything, /// because it did not: the server already has it. #[tokio::test] async fn a_row_the_server_accepted_cannot_be_cancelled() { let state = state().await; enqueue(&state, &op("create_group", "gone")).unwrap(); record_done(&state, "gone"); assert!(!cancel(&state, "gone").unwrap()); let queued = pending(&state).unwrap(); assert_eq!(queued.len(), 1); assert!(queued[0].done_at.is_some()); } /// The moment between "the server accepted it" and "a sync brought the group /// back" is real, and a section that showed neither would look like it lost /// the request. #[tokio::test] async fn a_done_row_survives_until_the_directory_catches_up() { let state = state().await; let conn = state.db.conn().unwrap(); synckit_client::store::directory::ensure_tables(&conn).unwrap(); drop(conn); enqueue(&state, &op("create_group", "landed")).unwrap(); record_done(&state, "landed"); drain_once(&state, 1).await; assert_eq!(pending(&state).unwrap().len(), 1, "the directory has not"); let mut conn = state.db.conn().unwrap(); synckit_client::store::directory::add_group( &mut conn, &synckit_client::store::directory::KnownGroup { id: synckit_client::GroupId::new(uuid::Uuid::from_u128(1)), name: "The Firm".to_owned(), gck_version: 1, is_admin: true, }, ) .unwrap(); drop(conn); drain_once(&state, 2).await; assert!( pending(&state).unwrap().is_empty(), "and now it has, so the row has nothing left to say" ); } #[tokio::test] async fn a_failure_is_stamped_with_its_reason_and_counted() { let state = state().await; enqueue(&state, &op("create_group", "bad")).unwrap(); record_failure(&state, "bad", "The server said no."); let queued = pending(&state).unwrap(); assert_eq!(queued[0].attempts, 1); assert_eq!(queued[0].last_error.as_deref(), Some("The server said no.")); assert!(queued[0].done_at.is_none(), "and it is still queued"); } }