Skip to main content

max / goingson

27.0 KB · 665 lines History Blame Raw
1 //! The group-admin queue, and the drainer that empties it.
2 //!
3 //! <!-- wiki: quasi-overview -->
4 //!
5 //! # Why the app has one
6 //!
7 //! `create_group`, `add_member` and `remove_member` each open a conversation
8 //! with a server. A described route handler is synchronous, by
9 //! `quasi_router`'s Decision 6, which exists so egui and a terminal need no
10 //! runtime. So Settings > Sharing could show the groups and could not change
11 //! them.
12 //!
13 //! Queueing is a local write, so it can be described. This is the outbox's
14 //! argument applied a second time, and Max's ruling behind that one ("have
15 //! GoingsOn use an outbox model explicitly", `a3c76a24`) is what makes it a
16 //! pattern rather than a workaround repeated. See [`crate::outbox`], which this
17 //! is deliberately shaped like: a reader who has read that one already knows how
18 //! this starts, stops, backs off and survives a failing tick.
19 //!
20 //! # What a queue buys here, which is not what it bought for mail
21 //!
22 //! There is no send-later value in creating a group. What there is:
23 //!
24 //! - the write survives being offline, instead of failing at the instant
25 //! somebody pressed the button. That matters more for these than for mail,
26 //! because they are rare and deliberate and nobody retries them by habit;
27 //! - a failure is a row with a reason on it, sitting in the section that caused
28 //! it, rather than a toast that has already gone. For `add_member` that is the
29 //! difference between "it did not work" and "it did not work, and here is
30 //! what the server said";
31 //! - `add_member` needs the master key loaded to seal the group key to the new
32 //! member. A queue turns "you cannot do this right now" into "this happens
33 //! once you unlock", which is honest and is not something a button could
34 //! offer.
35 //!
36 //! # Nothing here is described, and that is the design
37 //!
38 //! The description says "queue this". What drains the queue is not a screen and
39 //! has no address. That division is why a queue answers the async problem rather
40 //! than moving it: the async lives out here, where there has always been a
41 //! runtime.
42 //!
43 //! # What a failed attempt does
44 //!
45 //! Stamps the error, counts the attempt, and leaves the row queued, backing off
46 //! on the count exactly as the outbox does. A row that can never succeed sits
47 //! with its reason on it rather than being deleted, because an intention to add
48 //! somebody to a group that silently disappeared is worse than one still visible
49 //! and failing.
50
51 use std::sync::Arc;
52
53 use tauri::Manager;
54 use tokio::time::{Duration, interval};
55 use tokio_util::sync::CancellationToken;
56 use tracing::{debug, error, info};
57
58 use crate::state::{AppState, DESKTOP_USER_ID};
59
60 /// How often the drainer wakes.
61 ///
62 /// Matches [`crate::outbox`], and for the same reason: it is what makes "queue"
63 /// acceptable as the only way to act. A group is created within a minute of
64 /// being asked for, which is not immediate and is not a wait anybody watches.
65 const CHECK_INTERVAL_SECS: u64 = 60;
66
67 /// One queued admin write.
68 ///
69 /// [`Default`] rather than eight literal `None`s at every call site: there are
70 /// eight kinds now and no kind reads more than three of the payload columns, so
71 /// a construction that names only what it uses says which those are.
72 #[derive(Debug, Clone, Default)]
73 pub struct QueuedOp {
74 pub id: String,
75 pub kind: String,
76 pub group_id: Option<String>,
77 pub name: Option<String>,
78 pub email: Option<String>,
79 pub pubkey: Option<String>,
80 pub member_user_id: Option<String>,
81 /// The invitation `confirm_invite` and `revoke_invite` address.
82 pub invitation_id: Option<String>,
83 /// A code the user pasted, for `preview_invite` and `accept_invite`.
84 ///
85 /// Never an *issued* token: that one is returned once and is written
86 /// straight into the directory by the drainer. See migration 071.
87 pub invite_token: Option<String>,
88 /// How long an issued invitation should stand, for `create_invite`.
89 pub expires_in_hours: Option<i64>,
90 pub attempts: i32,
91 pub last_error: Option<String>,
92 pub done_at: Option<String>,
93 }
94
95 impl QueuedOp {
96 /// How this reads in the section that queued it.
97 ///
98 /// Written here rather than in the screen because the screen draws a row and
99 /// this is what the row says: a person who queued three of these wants to
100 /// know which is which, and the payload is the only thing that tells them
101 /// apart.
102 #[must_use]
103 pub fn describe(&self) -> String {
104 self.describe_confirming(None)
105 }
106
107 /// The same, with the fingerprint a queued confirm is authorizing.
108 ///
109 /// A confirm carries a group and an invitation id and deliberately carries
110 /// no fingerprint, because the drainer must re-read the server's answer
111 /// rather than act on a copy (migration 071). But "Confirm an invitation"
112 /// authorizes nothing legible: the fingerprint is the whole content of the
113 /// decision, and a queue row that does not name it asks the reader to take
114 /// the pending act on trust for the minute it sits there.
115 ///
116 /// So the screen resolves it from the directory as it draws and passes it
117 /// in. That is a display of the current answer rather than a second copy of
118 /// it, which is why it lives in the argument and not in the row.
119 #[must_use]
120 pub fn describe_confirming(&self, fingerprint: Option<&str>) -> String {
121 match self.kind.as_str() {
122 "create_group" => format!(
123 "Create the group {}",
124 self.name.as_deref().unwrap_or("(unnamed)")
125 ),
126 "add_member" => format!(
127 "Add {} to a group",
128 self.email.as_deref().unwrap_or("(no address)")
129 ),
130 "remove_member" => "Remove a member from a group".to_owned(),
131 "create_invite" => "Issue an invite code".to_owned(),
132 "revoke_invite" => "Cancel an invitation".to_owned(),
133 "confirm_invite" => fingerprint.map_or_else(
134 // The invitation has left `accepted` since it was queued, or
135 // the directory has not caught up. Both are honest reasons not
136 // to be able to name the fingerprint, and neither is a reason
137 // to name a stale one.
138 || "Admit somebody, once their fingerprint is checked".to_owned(),
139 |fingerprint| format!("Admit the holder of {fingerprint}"),
140 ),
141 "preview_invite" => "Read what an invite code leads to".to_owned(),
142 "accept_invite" => "Accept an invite code".to_owned(),
143 // A kind this build does not know, held rather than refused. It can
144 // only come from a newer build that queued it, and saying so is
145 // better than drawing a blank row.
146 other => format!("An action this version does not understand ({other})"),
147 }
148 }
149 }
150
151 /// Queue an admin write. The whole of what a described handler does.
152 pub fn enqueue(state: &AppState, op: &QueuedOp) -> Result<(), String> {
153 let conn = state.db.conn().map_err(|e| e.to_string())?;
154 conn.execute(
155 "INSERT INTO group_admin_queue \
156 (id, user_id, kind, group_id, name, email, pubkey, member_user_id, \
157 invitation_id, invite_token, expires_in_hours) \
158 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
159 rusqlite::params![
160 op.id,
161 DESKTOP_USER_ID.to_string(),
162 op.kind,
163 op.group_id,
164 op.name,
165 op.email,
166 op.pubkey,
167 op.member_user_id,
168 op.invitation_id,
169 op.invite_token,
170 op.expires_in_hours,
171 ],
172 )
173 .map_err(|e| e.to_string())?;
174 Ok(())
175 }
176
177 /// Everything still waiting, oldest first, plus anything done that the directory
178 /// has not caught up with yet.
179 ///
180 /// Both, because the moment between "the server accepted it" and "a sync brought
181 /// the group back" is real and a screen that showed neither would look like it
182 /// lost the request.
183 pub fn pending(state: &AppState) -> Result<Vec<QueuedOp>, String> {
184 let conn = state.db.conn().map_err(|e| e.to_string())?;
185 let mut stmt = conn
186 .prepare(
187 "SELECT id, kind, group_id, name, email, pubkey, member_user_id, \
188 invitation_id, invite_token, expires_in_hours, \
189 attempts, last_error, done_at \
190 FROM group_admin_queue WHERE user_id = ?1 ORDER BY queued_at",
191 )
192 .map_err(|e| e.to_string())?;
193 let rows = stmt
194 .query_map(rusqlite::params![DESKTOP_USER_ID.to_string()], |row| {
195 Ok(QueuedOp {
196 id: row.get(0)?,
197 kind: row.get(1)?,
198 group_id: row.get(2)?,
199 name: row.get(3)?,
200 email: row.get(4)?,
201 pubkey: row.get(5)?,
202 member_user_id: row.get(6)?,
203 invitation_id: row.get(7)?,
204 invite_token: row.get(8)?,
205 expires_in_hours: row.get(9)?,
206 attempts: row.get(10)?,
207 last_error: row.get(11)?,
208 done_at: row.get(12)?,
209 })
210 })
211 .map_err(|e| e.to_string())?
212 .collect::<Result<Vec<_>, _>>()
213 .map_err(|e| e.to_string())?;
214 Ok(rows)
215 }
216
217 /// Take a queued write back out.
218 ///
219 /// The way out of a row that will never succeed, and the reason a failure is
220 /// held rather than deleted: the person who queued it decides, not the drainer.
221 pub fn cancel(state: &AppState, id: &str) -> Result<bool, String> {
222 let conn = state.db.conn().map_err(|e| e.to_string())?;
223 let changed = conn
224 .execute(
225 "DELETE FROM group_admin_queue WHERE id = ?1 AND user_id = ?2 AND done_at IS NULL",
226 rusqlite::params![id, DESKTOP_USER_ID.to_string()],
227 )
228 .map_err(|e| e.to_string())?;
229 Ok(changed > 0)
230 }
231
232 /// Sweep the rows the server has accepted and the directory has caught up with.
233 ///
234 /// A done row is kept until the group it made is in the directory, so the
235 /// section can say "created" for the moment between the two. Once the directory
236 /// has it, the row has nothing left to say.
237 fn sweep_settled(state: &AppState) {
238 let Ok(conn) = state.db.conn() else { return };
239 // `create_group` is the only kind whose landing is observable in the
240 // directory. The other two change a member list, which is only fetched for
241 // groups this user administers and may legitimately not have refreshed yet,
242 // so they are swept on age instead.
243 let _ = conn.execute(
244 "DELETE FROM group_admin_queue \
245 WHERE done_at IS NOT NULL \
246 AND (kind = 'create_group' \
247 AND EXISTS (SELECT 1 FROM sync_groups WHERE name = group_admin_queue.name) \
248 OR done_at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-1 hour'))",
249 [],
250 );
251 }
252
253 fn record_failure(state: &AppState, id: &str, message: &str) {
254 let Ok(conn) = state.db.conn() else { return };
255 if let Err(error) = conn.execute(
256 "UPDATE group_admin_queue SET attempts = attempts + 1, last_error = ?2 WHERE id = ?1",
257 rusqlite::params![id, message],
258 ) {
259 error!("Group queue: could not record the failure: {error}");
260 }
261 }
262
263 fn record_done(state: &AppState, id: &str) {
264 let Ok(conn) = state.db.conn() else { return };
265 if let Err(error) = conn.execute(
266 "UPDATE group_admin_queue SET done_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), \
267 last_error = NULL WHERE id = ?1",
268 rusqlite::params![id],
269 ) {
270 error!("Group queue: could not record the success: {error}");
271 }
272 }
273
274 /// Start the drainer. Runs until cancelled.
275 pub async fn start_group_queue_drainer(app: tauri::AppHandle, cancel: CancellationToken) {
276 let mut check_interval = interval(Duration::from_secs(CHECK_INTERVAL_SECS));
277 let mut tick: u64 = 0;
278
279 info!("Group queue drainer started (checking every {CHECK_INTERVAL_SECS} seconds)");
280
281 loop {
282 tokio::select! {
283 () = cancel.cancelled() => {
284 info!("Group queue drainer shutting down");
285 break;
286 }
287 _ = check_interval.tick() => {}
288 }
289 tick = tick.wrapping_add(1);
290
291 let Some(state) = app.try_state::<Arc<AppState>>() else {
292 debug!("Group queue drainer: state not ready yet");
293 continue;
294 };
295 let state: Arc<AppState> = state.inner().clone();
296
297 drain_once(&state, tick).await;
298 }
299 }
300
301 /// One pass over the queue.
302 ///
303 /// Split from the loop so a test can run a pass without a Tauri handle or a
304 /// minute of waiting, exactly as [`crate::outbox::drain_once`] is.
305 pub async fn drain_once(state: &Arc<AppState>, tick: u64) {
306 sweep_settled(state);
307
308 let queued = match pending(state) {
309 Ok(queued) => queued,
310 Err(error) => {
311 error!("Group queue drainer: could not read the queue: {error}");
312 return;
313 }
314 };
315
316 for op in queued {
317 if op.done_at.is_some() {
318 continue;
319 }
320 // The outbox's curve, shared rather than re-derived: a row that has
321 // failed once is retried on the next wake, one that has failed six times
322 // every half hour or so, and the cap stops a long failure from becoming
323 // a silent drop wearing a backoff's clothes.
324 if !crate::outbox::due_on_tick(op.attempts, tick) {
325 continue;
326 }
327
328 // Not configured is not a failure to count: a device that has never
329 // signed in will never succeed at any of these, and counting attempts
330 // against it would back the row off to never while the reason stays the
331 // same. Left untouched, so it goes as soon as sync is set up.
332 let Some(client) = state.read_recovering() else {
333 continue;
334 };
335
336 // Opened per row rather than held across the loop: keeping a pooled
337 // connection in hand across an await takes it out of the pool for the
338 // length of a network call, which is the shape that starves a pool.
339 let mut conn = state.db.conn().ok();
340 let outcome = perform(&client, &op, conn.as_deref_mut()).await;
341 drop(conn);
342
343 match outcome {
344 Ok(()) => record_done(state, &op.id),
345 Err(message) => record_failure(state, &op.id, &message),
346 }
347 }
348 }
349
350 /// Perform one queued write against the server.
351 async fn perform(
352 client: &synckit_client::SyncKitClient,
353 op: &QueuedOp,
354 conn: Option<&mut rusqlite::Connection>,
355 ) -> Result<(), String> {
356 let group = |raw: &Option<String>| -> Result<synckit_client::GroupId, String> {
357 raw.as_deref()
358 .ok_or_else(|| "No group on the queued action.".to_owned())?
359 .parse::<uuid::Uuid>()
360 .map(synckit_client::GroupId::new)
361 .map_err(|_| "The queued action names a group id that will not parse.".to_owned())
362 };
363 let invitation = |raw: &Option<String>| -> Result<synckit_client::InvitationId, String> {
364 raw.as_deref()
365 .ok_or_else(|| "No invitation on the queued action.".to_owned())?
366 .parse::<uuid::Uuid>()
367 .map(synckit_client::InvitationId::new)
368 .map_err(|_| "The queued action names an invitation id that will not parse.".to_owned())
369 };
370 // Already normalised by the handler that queued it, deliberately: there is
371 // one accepted spelling of a pasted code and `normalize_invite_token` owns
372 // it. Re-normalising here would be a second place for the format to live.
373 //
374 // A function rather than a closure beside the two above, because it borrows
375 // out of its argument and a closure cannot state that the returned `&str`
376 // outlives the call.
377 fn token(raw: Option<&str>) -> Result<&str, String> {
378 raw.ok_or_else(|| "No invite code on the queued action.".to_owned())
379 }
380
381 match op.kind.as_str() {
382 "create_group" => {
383 let name = op
384 .name
385 .as_deref()
386 .ok_or_else(|| "No name on the queued action.".to_owned())?;
387 let group = client.create_group(name).await.map_err(|e| e.to_string())?;
388 // Into the directory here, the same as `commands::group::group_create`
389 // does on its own path: the group is real on the server, and this
390 // device would otherwise not know its name until the next cycle, so
391 // the section that queued it would show nothing landing.
392 if let Some(conn) = conn {
393 let known = synckit_client::store::directory::KnownGroup {
394 id: group.id,
395 name: group.name,
396 gck_version: group.gck_version,
397 is_admin: true,
398 };
399 if let Err(error) = synckit_client::store::directory::add_group(conn, &known) {
400 // Not a failure of the write: the group exists. The next
401 // cycle writes the whole directory anyway.
402 error!("Group queue: could not record the new group: {error}");
403 }
404 }
405 Ok(())
406 }
407 "add_member" => {
408 let email = op
409 .email
410 .as_deref()
411 .ok_or_else(|| "No address on the queued action.".to_owned())?;
412 let pubkey = op
413 .pubkey
414 .as_deref()
415 .ok_or_else(|| "No public key on the queued action.".to_owned())?;
416 client
417 .add_member(group(&op.group_id)?, email, pubkey)
418 .await
419 .map_err(|e| e.to_string())
420 }
421 "remove_member" => {
422 let member = op
423 .member_user_id
424 .as_deref()
425 .ok_or_else(|| "No member on the queued action.".to_owned())?
426 .parse::<uuid::Uuid>()
427 .map(synckit_client::UserId::new)
428 .map_err(|_| "The queued action names a user id that will not parse.".to_owned())?;
429 client
430 .remove_member(group(&op.group_id)?, member)
431 .await
432 .map_err(|e| e.to_string())
433 }
434 "create_invite" => {
435 let group = group(&op.group_id)?;
436 let invitation = client
437 .create_invitation(group, op.expires_in_hours)
438 .await
439 .map_err(|e| e.to_string())?;
440 // Not a nicety, unlike `create_group`'s `add_group` above. The
441 // server keeps only a hash of the token, so what came back is the
442 // only copy that will ever exist and this is the one moment it can
443 // be written down. A failure here has issued an invitation whose
444 // code nobody holds, which is why it is reported as a failure of
445 // the write rather than logged past.
446 let conn = conn.ok_or_else(|| {
447 "The invite was issued and could not be written down: no database \
448 connection. Revoke it from the group and issue another."
449 .to_owned()
450 })?;
451 synckit_client::store::directory::record_issued(conn, group, &invitation).map_err(
452 |error| {
453 format!(
454 "The invite was issued and could not be written down ({error}). \
455 Revoke it from the group and issue another."
456 )
457 },
458 )
459 }
460 "revoke_invite" => client
461 .revoke_invitation(group(&op.group_id)?, invitation(&op.invitation_id)?)
462 .await
463 .map_err(|e| e.to_string()),
464 // No fingerprint is passed, and none is stored. The queue row names the
465 // invitation and the server answers with the key it currently holds; a
466 // fingerprint copied at queue time would be authorizing a value nobody
467 // re-checked.
468 "confirm_invite" => client
469 .confirm_invitation(group(&op.group_id)?, invitation(&op.invitation_id)?, None)
470 .await
471 .map_err(|e| e.to_string()),
472 "preview_invite" => {
473 let token = token(op.invite_token.as_deref())?;
474 let preview = client
475 .preview_invitation(token)
476 .await
477 .map_err(|e| e.to_string())?;
478 let conn = conn.ok_or_else(|| {
479 "Read the code and could not write down the answer: no database connection."
480 .to_owned()
481 })?;
482 let known = synckit_client::store::directory::KnownPreview {
483 token: token.to_owned(),
484 group_name: preview.group_name,
485 inviter_email: preview.inviter_email,
486 redeemable: preview.redeemable,
487 state: preview.state,
488 expires_at: preview.expires_at.to_rfc3339(),
489 };
490 synckit_client::store::directory::write_preview(conn, &known)
491 .map_err(|error| format!("Could not write down what the code leads to: {error}"))
492 }
493 "accept_invite" => {
494 client
495 .accept_invitation(token(op.invite_token.as_deref())?)
496 .await
497 .map_err(|e| e.to_string())?;
498 // The code has done its whole job. Clearing it is what takes the
499 // preview section off the screen, so a failure here would leave an
500 // Accept control offering an act that already happened.
501 if let Some(conn) = conn
502 && let Err(error) = synckit_client::store::directory::clear_preview(conn)
503 {
504 error!("Group queue: could not forget the accepted code: {error}");
505 }
506 Ok(())
507 }
508 // Held rather than refused, and the message says why so it does not read
509 // as a bug. Only a newer build could have written it.
510 other => Err(format!(
511 "This version does not know how to perform `{other}`. It is kept, not lost."
512 )),
513 }
514 }
515
516 #[cfg(test)]
517 mod tests {
518 use super::*;
519
520 async fn state() -> Arc<AppState> {
521 let (state, _) = crate::test_utils::setup_test_state().await;
522 state
523 }
524
525 fn op(kind: &str, id: &str) -> QueuedOp {
526 QueuedOp {
527 id: id.to_owned(),
528 kind: kind.to_owned(),
529 group_id: Some("00000000-0000-0000-0000-000000000001".to_owned()),
530 name: Some("The Firm".to_owned()),
531 email: Some("them@localhost".to_owned()),
532 pubkey: Some("k".to_owned()),
533 member_user_id: Some("00000000-0000-0000-0000-000000000002".to_owned()),
534 ..Default::default()
535 }
536 }
537
538 /// A person who queued three of these wants to know which is which, and the
539 /// payload is the only thing that tells them apart.
540 #[test]
541 fn each_kind_says_what_it_will_do() {
542 assert_eq!(
543 op("create_group", "a").describe(),
544 "Create the group The Firm"
545 );
546 assert_eq!(
547 op("add_member", "a").describe(),
548 "Add them@localhost to a group"
549 );
550 assert_eq!(
551 op("remove_member", "a").describe(),
552 "Remove a member from a group"
553 );
554 }
555
556 /// Only a newer build could have written it, so saying so beats a blank row
557 /// and beats refusing to draw the queue at all.
558 #[test]
559 fn a_kind_from_a_newer_build_still_reads_as_something() {
560 let described = op("invite_member", "a").describe();
561 assert!(described.contains("does not understand"), "{described}");
562 assert!(described.contains("invite_member"), "{described}");
563 }
564
565 #[tokio::test]
566 async fn a_queued_action_comes_back_out_in_the_order_it_went_in() {
567 let state = state().await;
568 enqueue(&state, &op("create_group", "first")).unwrap();
569 enqueue(&state, &op("add_member", "second")).unwrap();
570
571 let queued = pending(&state).unwrap();
572 assert_eq!(queued.len(), 2);
573 assert_eq!(queued[0].id, "first");
574 assert_eq!(queued[1].id, "second");
575 }
576
577 /// A device that has never signed in will never succeed at any of these, and
578 /// counting attempts against it would back the row off to never while the
579 /// reason stays the same. It has to go as soon as sync is set up.
580 #[tokio::test]
581 async fn a_pass_with_no_client_leaves_the_row_untouched() {
582 let state = state().await;
583 enqueue(&state, &op("create_group", "waiting")).unwrap();
584
585 drain_once(&state, 1).await;
586
587 let queued = pending(&state).unwrap();
588 assert_eq!(queued.len(), 1, "still queued");
589 assert_eq!(queued[0].attempts, 0, "and not counted against");
590 assert!(queued[0].last_error.is_none());
591 }
592
593 #[tokio::test]
594 async fn cancelling_takes_it_out_and_says_whether_it_did() {
595 let state = state().await;
596 enqueue(&state, &op("create_group", "mistake")).unwrap();
597
598 assert!(cancel(&state, "mistake").unwrap());
599 assert!(pending(&state).unwrap().is_empty());
600 assert!(!cancel(&state, "mistake").unwrap(), "twice is not a lie");
601 }
602
603 /// A cancel that raced the drainer must not report that it undid anything,
604 /// because it did not: the server already has it.
605 #[tokio::test]
606 async fn a_row_the_server_accepted_cannot_be_cancelled() {
607 let state = state().await;
608 enqueue(&state, &op("create_group", "gone")).unwrap();
609 record_done(&state, "gone");
610
611 assert!(!cancel(&state, "gone").unwrap());
612 let queued = pending(&state).unwrap();
613 assert_eq!(queued.len(), 1);
614 assert!(queued[0].done_at.is_some());
615 }
616
617 /// The moment between "the server accepted it" and "a sync brought the group
618 /// back" is real, and a section that showed neither would look like it lost
619 /// the request.
620 #[tokio::test]
621 async fn a_done_row_survives_until_the_directory_catches_up() {
622 let state = state().await;
623 let conn = state.db.conn().unwrap();
624 synckit_client::store::directory::ensure_tables(&conn).unwrap();
625 drop(conn);
626
627 enqueue(&state, &op("create_group", "landed")).unwrap();
628 record_done(&state, "landed");
629
630 drain_once(&state, 1).await;
631 assert_eq!(pending(&state).unwrap().len(), 1, "the directory has not");
632
633 let mut conn = state.db.conn().unwrap();
634 synckit_client::store::directory::add_group(
635 &mut conn,
636 &synckit_client::store::directory::KnownGroup {
637 id: synckit_client::GroupId::new(uuid::Uuid::from_u128(1)),
638 name: "The Firm".to_owned(),
639 gck_version: 1,
640 is_admin: true,
641 },
642 )
643 .unwrap();
644 drop(conn);
645
646 drain_once(&state, 2).await;
647 assert!(
648 pending(&state).unwrap().is_empty(),
649 "and now it has, so the row has nothing left to say"
650 );
651 }
652
653 #[tokio::test]
654 async fn a_failure_is_stamped_with_its_reason_and_counted() {
655 let state = state().await;
656 enqueue(&state, &op("create_group", "bad")).unwrap();
657 record_failure(&state, "bad", "The server said no.");
658
659 let queued = pending(&state).unwrap();
660 assert_eq!(queued[0].attempts, 1);
661 assert_eq!(queued[0].last_error.as_deref(), Some("The server said no."));
662 assert!(queued[0].done_at.is_none(), "and it is still queued");
663 }
664 }
665