agent_effects_store/
testkit.rs

1//! Conformance suite for [`EffectStore`] implementations.
2//!
3//! Enabled by the `testkit` feature. Add it as a dev-dependency feature:
4//!
5//! ```toml
6//! [dev-dependencies]
7//! agent-effects-store = { version = "0.1", features = ["testkit"] }
8//! ```
9
10use std::future::Future;
11use std::sync::Arc;
12use std::time::{Duration, SystemTime};
13
14use serde_json::json;
15
16use crate::failure::FailureClass;
17use crate::id::{EffectId, EffectKey, EffectName, LogicalKey, WorkerId};
18use crate::kind::EffectKind;
19use crate::state::{ALL_STATUSES, ALL_TRANSITIONS, EffectStatus, Transition};
20use crate::{
21    EffectEvent, EffectRecord, EffectStore, ErrorRecord, Lease, ListQuery, NewEffect, PruneQuery,
22    StoreError, TransitionRequest,
23};
24
25/// Runs the store conformance suite against stores built by `make_store`.
26///
27/// Every case gets a fresh store. A failing case panics. Some cases spawn
28/// tasks, so call this from inside a Tokio runtime, ideally a multi-threaded
29/// one so races are real.
30///
31/// ```ignore
32/// #[tokio::test(flavor = "multi_thread")]
33/// async fn conformance() {
34///     agent_effects_store::testkit::conformance(|| async { MyStore::open_temp().await }).await;
35/// }
36/// ```
37pub async fn conformance<S, F, Fut>(make_store: F)
38where
39    S: EffectStore,
40    F: Fn() -> Fut,
41    Fut: Future<Output = S>,
42{
43    insert_and_read_back(make_store().await).await;
44    supplementary_evidence_never_changes_authority(make_store().await).await;
45    insert_is_idempotent_per_key(make_store().await).await;
46    concurrent_inserts_converge(make_store().await).await;
47    missing_records_are_reported(make_store().await).await;
48    leases_are_exclusive(make_store().await).await;
49    takeover_fences_the_old_owner(make_store().await).await;
50    renewal_extends_and_is_strict(make_store().await).await;
51    release_frees_the_lease(make_store().await).await;
52    transitions_persist_with_events(make_store().await).await;
53    rejected_transitions_change_nothing(make_store().await).await;
54    unleased_transitions_for_operators(make_store().await).await;
55    terminal_records_are_final(make_store().await).await;
56    listing_filters_and_pages(make_store().await).await;
57    doubt_is_persisted_and_guards_failure(make_store().await).await;
58    a_new_attempt_clears_the_output(make_store().await).await;
59    compensation_is_persisted(make_store().await).await;
60    approval_is_persisted(make_store().await).await;
61    pruning_removes_only_settled_idle_old_records(make_store().await).await;
62}
63
64async fn supplementary_evidence_never_changes_authority<S: EffectStore>(store: S) {
65    use crate::{LateObservation, LateOutcome};
66    let record = insert(&store, 1).await;
67    let old = store
68        .acquire_lease(record.id, &worker("old"), t(0), Duration::from_secs(1))
69        .await
70        .unwrap();
71    let executing = drive(
72        &store,
73        record,
74        Some(&old),
75        &[Transition::StartAttempt],
76        t(0),
77    )
78    .await;
79    let new = store
80        .acquire_lease(executing.id, &worker("new"), t(2), Duration::from_secs(30))
81        .await
82        .unwrap();
83    let before = reload(&store, executing.id).await;
84    let history = store.events(executing.id).await.unwrap();
85    let first = LateObservation::new(&executing, &old, LateOutcome::Succeeded, t(3));
86    let second = LateObservation::new(&executing, &old, LateOutcome::Failed, t(4));
87    store.append_late_observation(first.clone()).await.unwrap();
88    let mut conflicting = first.clone();
89    conflicting.outcome = LateOutcome::Failed;
90    store.append_late_observation(conflicting).await.unwrap();
91    store.append_late_observation(second.clone()).await.unwrap();
92    assert_eq!(
93        store.late_observations(before.id, None, 1).await.unwrap(),
94        std::slice::from_ref(&first)
95    );
96    assert_eq!(
97        store
98            .late_observations(before.id, Some(first.id), 10)
99            .await
100            .unwrap(),
101        [second]
102    );
103    assert_eq!(
104        store.late_observations(before.id, None, 0).await.unwrap(),
105        [] as [LateObservation; 0]
106    );
107    assert_eq!(reload(&store, before.id).await, before);
108    assert_eq!(store.events(before.id).await.unwrap(), history);
109    // Even evidence of success does not allow the stale worker to commit.
110    assert!(matches!(
111        store
112            .transition(TransitionRequest::new(
113                &before,
114                Some(&old),
115                Transition::Succeeded,
116                t(3)
117            ))
118            .await,
119        Err(StoreError::LeaseLost)
120    ));
121    let settled = drive(
122        &store,
123        before,
124        Some(&new),
125        &[
126            Transition::LeaseExpired,
127            Transition::StartVerification,
128            Transition::VerificationConfirmed,
129        ],
130        t(3),
131    )
132    .await;
133    store.release_lease(&new).await.unwrap();
134    assert_eq!(
135        store
136            .prune(PruneQuery::new(
137                EffectStatus::Committed,
138                Duration::from_secs(1),
139                t(100)
140            ))
141            .await
142            .unwrap(),
143        1
144    );
145    assert!(store.get(settled.id).await.unwrap().is_none());
146    assert!(matches!(
147        store.late_observations(settled.id, None, 10).await,
148        Err(StoreError::NotFound(_))
149    ));
150}
151
152const TTL: Duration = Duration::from_secs(30);
153
154/// Test time `secs` seconds after a fixed base, at millisecond precision.
155fn t(secs: u64) -> SystemTime {
156    SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000 + secs)
157}
158
159fn key(n: u32) -> EffectKey {
160    EffectKey::new(
161        EffectName::new("testkit.effect").unwrap(),
162        LogicalKey::new(format!("key-{n}")).unwrap(),
163    )
164}
165
166fn new_effect(n: u32) -> NewEffect {
167    NewEffect {
168        input_schema_version: 7,
169        input: Some(json!({ "n": n, "nested": { "b": [1, 2], "a": null } })),
170        input_fingerprint: Some(format!("fingerprint-{n}")),
171        created_by: Some("agent:testkit".into()),
172        execution_contract: Some(crate::ExecutionContract {
173            remote_idempotency: true,
174            handler_version: "test-v2".into(),
175            adapter: Some("test-provider".into()),
176            verification: crate::VerificationContract::EventuallyConsistent {
177                settle: Duration::from_secs(17),
178            },
179            ..crate::ExecutionContract::new(EffectKind::IrreversibleWrite, 7)
180        }),
181        ..NewEffect::new(key(n), EffectKind::IrreversibleWrite, t(0))
182    }
183}
184
185fn worker(name: &str) -> WorkerId {
186    WorkerId::new(name)
187}
188
189async fn insert<S: EffectStore>(store: &S, n: u32) -> EffectRecord {
190    store.insert_or_get(new_effect(n)).await.unwrap().record
191}
192
193async fn reload<S: EffectStore>(store: &S, id: EffectId) -> EffectRecord {
194    store.get(id).await.unwrap().expect("record exists")
195}
196
197/// Applies `transitions` in order under `lease` at time `now`.
198async fn drive<S: EffectStore>(
199    store: &S,
200    mut record: EffectRecord,
201    lease: Option<&Lease>,
202    transitions: &[Transition],
203    now: SystemTime,
204) -> EffectRecord {
205    for &transition in transitions {
206        record = store
207            .transition(TransitionRequest::new(&record, lease, transition, now))
208            .await
209            .unwrap_or_else(|e| panic!("{transition} from {}: {e}", record.status));
210    }
211    record
212}
213
214async fn insert_and_read_back<S: EffectStore>(store: S) {
215    let new = new_effect(1);
216    let outcome = store.insert_or_get(new.clone()).await.unwrap();
217    assert!(outcome.inserted);
218    let record = outcome.record;
219    assert_eq!(record.id, new.id);
220    assert_eq!(record.key, new.key);
221    assert_eq!(record.kind, new.kind);
222    assert_eq!(record.status, EffectStatus::Pending);
223    assert_eq!(record.input, new.input);
224    assert_eq!(record.input_fingerprint, new.input_fingerprint);
225    assert_eq!(record.execution_contract, new.execution_contract);
226    assert_eq!(record.input_schema_version, 7);
227    assert_eq!(record.output_schema_version, 1);
228    assert_eq!(record.created_by, new.created_by);
229    assert_eq!(
230        (record.version, record.lease_epoch, record.attempt_count),
231        (0, 0, 0)
232    );
233    assert_eq!((record.created_at, record.updated_at), (t(0), t(0)));
234    assert_eq!(record.lease_owner, None);
235
236    assert_eq!(store.get(record.id).await.unwrap(), Some(record.clone()));
237    assert_eq!(
238        store.get_by_key(&record.key).await.unwrap(),
239        Some(record.clone())
240    );
241    assert_eq!(
242        store.events(record.id).await.unwrap(),
243        Vec::<EffectEvent>::new()
244    );
245}
246
247async fn insert_is_idempotent_per_key<S: EffectStore>(store: S) {
248    let first = insert(&store, 1).await;
249    let again = NewEffect {
250        input: Some(json!("different")),
251        ..NewEffect::new(key(1), EffectKind::Read, t(5))
252    };
253    let outcome = store.insert_or_get(again).await.unwrap();
254    assert!(
255        !outcome.inserted,
256        "a second insert for the same key must not insert"
257    );
258    assert_eq!(
259        outcome.record, first,
260        "the existing record must come back untouched"
261    );
262
263    let other = insert(&store, 2).await;
264    assert_ne!(other.id, first.id);
265}
266
267async fn concurrent_inserts_converge<S: EffectStore>(store: S) {
268    let store = Arc::new(store);
269    let tasks: Vec<_> = (0..16)
270        .map(|_| {
271            let store = Arc::clone(&store);
272            tokio::spawn(async move { store.insert_or_get(new_effect(7)).await.unwrap() })
273        })
274        .collect();
275    let mut outcomes = Vec::new();
276    for task in tasks {
277        outcomes.push(task.await.unwrap());
278    }
279    let inserted = outcomes.iter().filter(|o| o.inserted).count();
280    assert_eq!(inserted, 1, "exactly one concurrent insert may win");
281    let id = outcomes[0].record.id;
282    assert!(
283        outcomes.iter().all(|o| o.record.id == id),
284        "all callers must see one record"
285    );
286}
287
288async fn missing_records_are_reported<S: EffectStore>(store: S) {
289    let id = EffectId::new();
290    assert_eq!(store.get(id).await.unwrap(), None);
291    assert_eq!(store.get_by_key(&key(404)).await.unwrap(), None);
292    assert!(matches!(
293        store.acquire_lease(id, &worker("a"), t(0), TTL).await,
294        Err(StoreError::NotFound(missing)) if missing == id
295    ));
296    let mut request = TransitionRequest::new(
297        &insert(&store, 1).await,
298        None,
299        Transition::StartAttempt,
300        t(0),
301    );
302    request.id = id;
303    assert!(matches!(
304        store.transition(request).await,
305        Err(StoreError::NotFound(_))
306    ));
307    assert!(matches!(
308        store.events(id).await,
309        Err(StoreError::NotFound(_))
310    ));
311}
312
313async fn leases_are_exclusive<S: EffectStore>(store: S) {
314    let record = insert(&store, 1).await;
315    let lease = store
316        .acquire_lease(record.id, &worker("a"), t(0), TTL)
317        .await
318        .unwrap();
319    assert_eq!((lease.epoch, lease.expires_at), (1, t(30)));
320    assert_eq!(lease.owner, worker("a"));
321
322    for contender in ["b", "a"] {
323        match store
324            .acquire_lease(record.id, &worker(contender), t(1), TTL)
325            .await
326        {
327            Err(StoreError::LeaseHeld { owner, expires_at }) => {
328                assert_eq!(owner, worker("a"));
329                assert_eq!(expires_at, t(30));
330            }
331            other => panic!("{contender} acquired a held lease: {other:?}"),
332        }
333    }
334}
335
336async fn takeover_fences_the_old_owner<S: EffectStore>(store: S) {
337    let record = insert(&store, 1).await;
338    let old = store
339        .acquire_lease(record.id, &worker("a"), t(0), TTL)
340        .await
341        .unwrap();
342    assert!(matches!(
343        store
344            .acquire_lease(record.id, &worker("b"), t(29), TTL)
345            .await,
346        Err(StoreError::LeaseHeld { .. })
347    ));
348    let new = store
349        .acquire_lease(record.id, &worker("b"), t(30), TTL)
350        .await
351        .unwrap();
352    assert_eq!(new.epoch, 2);
353
354    let stale = TransitionRequest::new(&record, Some(&old), Transition::StartAttempt, t(31));
355    assert!(matches!(
356        store.transition(stale).await,
357        Err(StoreError::LeaseLost)
358    ));
359    assert!(matches!(
360        store.renew_lease(&old, t(31), TTL).await,
361        Err(StoreError::LeaseLost)
362    ));
363    store.release_lease(&old).await.unwrap();
364
365    let current = reload(&store, record.id).await;
366    assert_eq!(
367        current.lease_owner,
368        Some(worker("b")),
369        "a stale release must not free b's lease"
370    );
371    assert_eq!((current.lease_epoch, current.version), (2, 0));
372}
373
374async fn renewal_extends_and_is_strict<S: EffectStore>(store: S) {
375    let record = insert(&store, 1).await;
376    let lease = store
377        .acquire_lease(record.id, &worker("a"), t(0), TTL)
378        .await
379        .unwrap();
380    let renewed = store.renew_lease(&lease, t(20), TTL).await.unwrap();
381    assert_eq!((renewed.epoch, renewed.expires_at), (lease.epoch, t(50)));
382    assert!(matches!(
383        store
384            .acquire_lease(record.id, &worker("b"), t(40), TTL)
385            .await,
386        Err(StoreError::LeaseHeld { .. })
387    ));
388    assert!(
389        matches!(
390            store.renew_lease(&renewed, t(50), TTL).await,
391            Err(StoreError::LeaseLost)
392        ),
393        "an expired lease must not be revived"
394    );
395    assert_eq!(
396        reload(&store, record.id).await.version,
397        0,
398        "lease operations must not bump the version"
399    );
400}
401
402async fn release_frees_the_lease<S: EffectStore>(store: S) {
403    let record = insert(&store, 1).await;
404    let lease = store
405        .acquire_lease(record.id, &worker("a"), t(0), TTL)
406        .await
407        .unwrap();
408    store.release_lease(&lease).await.unwrap();
409    store.release_lease(&lease).await.unwrap();
410    let next = store
411        .acquire_lease(record.id, &worker("b"), t(1), TTL)
412        .await
413        .unwrap();
414    assert_eq!(next.epoch, 2);
415}
416
417async fn transitions_persist_with_events<St: EffectStore>(store: St) {
418    use EffectStatus as S;
419    use Transition as T;
420
421    let record = insert(&store, 1).await;
422    let lease = store
423        .acquire_lease(record.id, &worker("a"), t(0), TTL)
424        .await
425        .unwrap();
426    let record = drive(
427        &store,
428        record,
429        Some(&lease),
430        &[Transition::StartAttempt],
431        t(1),
432    )
433    .await;
434    assert_eq!(record.status, EffectStatus::Executing);
435    assert_eq!(record.attempt_started_at, Some(t(1)));
436
437    let error = ErrorRecord {
438        class: Some(FailureClass::RateLimited {
439            retry_after: Some(Duration::from_secs(9)),
440        }),
441        message: "429".into(),
442    };
443    let mut retry = TransitionRequest::new(&record, Some(&lease), Transition::ScheduleRetry, t(2));
444    retry.next_attempt_at = Some(t(11));
445    retry.error = Some(error.clone());
446    retry.actor = Some("worker:a".into());
447    retry.payload = Some(json!({ "delay_ms": 9000 }));
448    let record = store.transition(retry).await.unwrap();
449    assert_eq!(
450        record,
451        reload(&store, record.id).await,
452        "transition must return the stored record"
453    );
454    assert_eq!(record.next_attempt_at, Some(t(11)));
455    assert_eq!(record.attempt_ended_at, Some(t(2)));
456    assert_eq!(record.last_error, Some(error));
457
458    let record = drive(
459        &store,
460        record,
461        Some(&lease),
462        &[Transition::StartAttempt],
463        t(11),
464    )
465    .await;
466    let output = json!({ "payment_id": "pi_123", "amount": 4200 });
467    let mut verify =
468        TransitionRequest::new(&record, Some(&lease), Transition::StartVerification, t(12));
469    verify.output = Some(output.clone());
470    verify.output_schema_version = Some(11);
471    let record = store.transition(verify).await.unwrap();
472    let record = drive(
473        &store,
474        record,
475        Some(&lease),
476        &[Transition::VerificationConfirmed],
477        t(13),
478    )
479    .await;
480
481    let stored = reload(&store, record.id).await;
482    assert_eq!(stored.status, EffectStatus::Committed);
483    assert_eq!(stored.output, Some(output));
484    assert_eq!(stored.output_schema_version, 11);
485    assert_eq!(stored.committed_at, Some(t(13)));
486    assert_eq!((stored.attempt_count, stored.version), (2, 5));
487    assert_eq!(stored.next_attempt_at, None);
488    assert_eq!(stored.attempt_ended_at, Some(t(12)));
489
490    let events = store.events(record.id).await.unwrap();
491    let trail: Vec<_> = events
492        .iter()
493        .map(|e| (e.sequence, e.transition, e.from, e.to, e.attempt))
494        .collect();
495    assert_eq!(
496        trail,
497        [
498            (1, T::StartAttempt, S::Pending, S::Executing, 1),
499            (2, T::ScheduleRetry, S::Executing, S::Pending, 1),
500            (3, T::StartAttempt, S::Pending, S::Executing, 2),
501            (4, T::StartVerification, S::Executing, S::Verifying, 2),
502            (5, T::VerificationConfirmed, S::Verifying, S::Committed, 2),
503        ]
504    );
505    assert_eq!(events[1].actor.as_deref(), Some("worker:a"));
506    assert_eq!(events[1].payload, Some(json!({ "delay_ms": 9000 })));
507    assert_eq!(events[1].at, t(2));
508}
509
510async fn rejected_transitions_change_nothing<S: EffectStore>(store: S) {
511    let record = insert(&store, 1).await;
512    let lease = store
513        .acquire_lease(record.id, &worker("a"), t(0), TTL)
514        .await
515        .unwrap();
516    let record = reload(&store, record.id).await;
517
518    let illegal = TransitionRequest::new(&record, Some(&lease), Transition::Succeeded, t(1));
519    let stale_version = TransitionRequest {
520        expected_version: 7,
521        ..TransitionRequest::new(&record, Some(&lease), Transition::StartAttempt, t(1))
522    };
523    let unleased = TransitionRequest::new(&record, None, Transition::StartAttempt, t(1));
524    let expired = TransitionRequest::new(&record, Some(&lease), Transition::StartAttempt, t(30));
525
526    assert!(matches!(
527        store.transition(illegal).await,
528        Err(StoreError::InvalidTransition(_))
529    ));
530    assert!(matches!(
531        store.transition(stale_version).await,
532        Err(StoreError::VersionConflict {
533            expected: 7,
534            actual: 0
535        })
536    ));
537    assert!(matches!(
538        store.transition(unleased).await,
539        Err(StoreError::LeaseHeld { .. })
540    ));
541    assert!(matches!(
542        store.transition(expired).await,
543        Err(StoreError::LeaseLost)
544    ));
545
546    assert_eq!(reload(&store, record.id).await, record);
547    assert_eq!(
548        store.events(record.id).await.unwrap(),
549        Vec::<EffectEvent>::new()
550    );
551}
552
553async fn unleased_transitions_for_operators<S: EffectStore>(store: S) {
554    let record = insert(&store, 1).await;
555    let lease = store
556        .acquire_lease(record.id, &worker("a"), t(0), TTL)
557        .await
558        .unwrap();
559    let record = drive(
560        &store,
561        record,
562        Some(&lease),
563        &[Transition::StartAttempt, Transition::OutcomeUnknown],
564        t(1),
565    )
566    .await;
567    store.release_lease(&lease).await.unwrap();
568
569    let record = drive(&store, record, None, &[Transition::Escalate], t(2)).await;
570    assert_eq!(record.status, EffectStatus::NeedsIntervention);
571    let mut resolve = TransitionRequest::new(&record, None, Transition::ResolvedApplied, t(3));
572    resolve.output = Some(json!("confirmed by operator"));
573    resolve.actor = Some("operator:dennis".into());
574    let record = store.transition(resolve).await.unwrap();
575    assert_eq!(record.status, EffectStatus::Committed);
576    assert_eq!(record.output, Some(json!("confirmed by operator")));
577}
578
579async fn terminal_records_are_final<S: EffectStore>(store: S) {
580    let record = insert(&store, 1).await;
581    let lease = store
582        .acquire_lease(record.id, &worker("a"), t(0), TTL)
583        .await
584        .unwrap();
585    let record = drive(
586        &store,
587        record,
588        Some(&lease),
589        &[Transition::StartAttempt, Transition::Succeeded],
590        t(1),
591    )
592    .await;
593    store.release_lease(&lease).await.unwrap();
594    let record = reload(&store, record.id).await;
595    // Committed has one exit, compensation; nothing else may leave it.
596    for transition in ALL_TRANSITIONS
597        .into_iter()
598        .filter(|t| *t != Transition::StartCompensation)
599    {
600        let request = TransitionRequest::new(&record, None, transition, t(2));
601        assert!(
602            matches!(
603                store.transition(request).await,
604                Err(StoreError::InvalidTransition(_))
605            ),
606            "{transition} left a committed record"
607        );
608    }
609    assert_eq!(reload(&store, record.id).await, record);
610
611    // Compensated is terminal.
612    let record = drive(
613        &store,
614        record,
615        None,
616        &[
617            Transition::StartCompensation,
618            Transition::CompensationSucceeded,
619        ],
620        t(3),
621    )
622    .await;
623    for transition in ALL_TRANSITIONS {
624        let request = TransitionRequest::new(&record, None, transition, t(4));
625        assert!(
626            matches!(
627                store.transition(request).await,
628                Err(StoreError::InvalidTransition(_))
629            ),
630            "{transition} left a compensated record"
631        );
632    }
633}
634
635async fn listing_filters_and_pages<S: EffectStore>(store: S) {
636    let now = t(100);
637    let mut ids = Vec::new();
638    for n in 0..5 {
639        ids.push(insert(&store, n).await.id);
640    }
641    // 0: Pending, no lease.
642    // 1: Executing, live lease.
643    let lease = store
644        .acquire_lease(ids[1], &worker("a"), t(90), TTL)
645        .await
646        .unwrap();
647    drive(
648        &store,
649        reload(&store, ids[1]).await,
650        Some(&lease),
651        &[Transition::StartAttempt],
652        t(90),
653    )
654    .await;
655    // 2: Executing, lease expired.
656    let lease = store
657        .acquire_lease(ids[2], &worker("a"), t(0), TTL)
658        .await
659        .unwrap();
660    drive(
661        &store,
662        reload(&store, ids[2]).await,
663        Some(&lease),
664        &[Transition::StartAttempt],
665        t(1),
666    )
667    .await;
668    // 3: Verifying, lease released.
669    let lease = store
670        .acquire_lease(ids[3], &worker("a"), t(0), TTL)
671        .await
672        .unwrap();
673    drive(
674        &store,
675        reload(&store, ids[3]).await,
676        Some(&lease),
677        &[Transition::StartAttempt, Transition::StartVerification],
678        t(1),
679    )
680    .await;
681    store.release_lease(&lease).await.unwrap();
682    // 4: Unknown.
683    let lease = store
684        .acquire_lease(ids[4], &worker("a"), t(0), TTL)
685        .await
686        .unwrap();
687    drive(
688        &store,
689        reload(&store, ids[4]).await,
690        Some(&lease),
691        &[Transition::StartAttempt, Transition::OutcomeUnknown],
692        t(1),
693    )
694    .await;
695
696    let listed = |query: ListQuery| {
697        let store = &store;
698        async move {
699            store
700                .list(query)
701                .await
702                .unwrap()
703                .into_iter()
704                .map(|r| r.id)
705                .collect::<Vec<_>>()
706        }
707    };
708    assert_eq!(
709        listed(ListQuery::statuses([EffectStatus::Pending])).await,
710        [ids[0]]
711    );
712    assert_eq!(
713        listed(ListQuery::expired_leases(now)).await,
714        [ids[2], ids[3]]
715    );
716    assert_eq!(
717        listed(ListQuery::statuses([]).limit(2)).await,
718        [ids[0], ids[1]]
719    );
720    assert_eq!(
721        listed(ListQuery::statuses([]).after(ids[1])).await,
722        [ids[2], ids[3], ids[4]]
723    );
724    assert_eq!(
725        listed(ListQuery::expired_leases(now).after(ids[2])).await,
726        [ids[3]]
727    );
728
729    let records = store
730        .list(ListQuery::statuses([EffectStatus::Unknown]))
731        .await
732        .unwrap();
733    assert_eq!(
734        records,
735        [reload(&store, ids[4]).await],
736        "listed records must be complete"
737    );
738}
739
740async fn doubt_is_persisted_and_guards_failure<S: EffectStore>(store: S) {
741    let record = insert(&store, 1).await;
742    assert!(!record.may_have_applied);
743    let lease = store
744        .acquire_lease(record.id, &worker("a"), t(0), TTL)
745        .await
746        .unwrap();
747    let record = drive(
748        &store,
749        record,
750        Some(&lease),
751        &[
752            Transition::StartAttempt,
753            Transition::OutcomeUnknown,
754            Transition::ScheduleRetry,
755            Transition::StartAttempt,
756        ],
757        t(1),
758    )
759    .await;
760    assert!(
761        reload(&store, record.id).await.may_have_applied,
762        "the store must persist may_have_applied"
763    );
764    let fail = TransitionRequest::new(&record, Some(&lease), Transition::FailedDefinitively, t(2));
765    assert!(
766        matches!(
767            store.transition(fail).await,
768            Err(StoreError::InvalidTransition(_))
769        ),
770        "a failed attempt must not fail an effect that may have applied"
771    );
772    assert_eq!(reload(&store, record.id).await, record);
773}
774
775async fn a_new_attempt_clears_the_output<S: EffectStore>(store: S) {
776    let record = insert(&store, 1).await;
777    let lease = store
778        .acquire_lease(record.id, &worker("a"), t(0), TTL)
779        .await
780        .unwrap();
781    let record = drive(
782        &store,
783        record,
784        Some(&lease),
785        &[Transition::StartAttempt],
786        t(1),
787    )
788    .await;
789    let mut verify =
790        TransitionRequest::new(&record, Some(&lease), Transition::StartVerification, t(1));
791    verify.output = Some(json!({ "attempt": 1 }));
792    let record = store.transition(verify).await.unwrap();
793    assert_eq!(
794        reload(&store, record.id).await.output,
795        Some(json!({ "attempt": 1 }))
796    );
797    // Shown not to have applied; the next attempt must not inherit its
798    // output.
799    let record = drive(
800        &store,
801        record,
802        Some(&lease),
803        &[Transition::ScheduleRetry, Transition::StartAttempt],
804        t(2),
805    )
806    .await;
807    assert_eq!(record.output, None);
808    assert_eq!(
809        reload(&store, record.id).await.output,
810        None,
811        "the store must persist the cleared output"
812    );
813}
814
815async fn compensation_is_persisted<S: EffectStore>(store: S) {
816    let record = insert(&store, 1).await;
817    let lease = store
818        .acquire_lease(record.id, &worker("a"), t(0), TTL)
819        .await
820        .unwrap();
821    let record = drive(
822        &store,
823        record,
824        Some(&lease),
825        &[
826            Transition::StartAttempt,
827            Transition::Succeeded,
828            Transition::StartCompensation,
829        ],
830        t(1),
831    )
832    .await;
833    let mut retry = TransitionRequest::new(
834        &record,
835        Some(&lease),
836        Transition::ScheduleCompensationRetry,
837        t(2),
838    );
839    retry.next_attempt_at = Some(t(5));
840    let record = store.transition(retry).await.unwrap();
841    let stored = reload(&store, record.id).await;
842    assert_eq!(stored, record, "transition must return the stored record");
843    assert_eq!(stored.status, EffectStatus::Compensating);
844    assert_eq!(
845        stored.compensation_attempts, 1,
846        "a scheduled retry is not an attempt until it starts"
847    );
848    assert_eq!(stored.next_attempt_at, Some(t(5)));
849
850    let record = drive(
851        &store,
852        record,
853        Some(&lease),
854        &[Transition::StartCompensationRetry],
855        t(5),
856    )
857    .await;
858    let stored = reload(&store, record.id).await;
859    assert_eq!(stored.compensation_attempts, 2);
860    assert_eq!(stored.next_attempt_at, None);
861
862    let record = drive(
863        &store,
864        record,
865        Some(&lease),
866        &[Transition::CompensationFailed],
867        t(6),
868    )
869    .await;
870    store.release_lease(&lease).await.unwrap();
871    let record = drive(
872        &store,
873        reload(&store, record.id).await,
874        None,
875        &[Transition::ResolvedRetry],
876        t(7),
877    )
878    .await;
879    assert_eq!(
880        record.status,
881        EffectStatus::Compensating,
882        "an operator can retry it"
883    );
884    let listed = store
885        .list(ListQuery::statuses([EffectStatus::Compensating]))
886        .await
887        .unwrap();
888    assert_eq!(listed, [record]);
889}
890
891async fn approval_is_persisted<S: EffectStore>(store: S) {
892    let record = insert(&store, 1).await;
893    assert!(!record.approved);
894    let record = drive(&store, record, None, &[Transition::RequestApproval], t(1)).await;
895    assert_eq!(record.status, EffectStatus::AwaitingApproval);
896    let mut approve = TransitionRequest::new(&record, None, Transition::Approve, t(2));
897    approve.actor = Some("operator:dennis".into());
898    let record = store.transition(approve).await.unwrap();
899    let stored = reload(&store, record.id).await;
900    assert_eq!(stored, record, "transition must return the stored record");
901    assert_eq!(stored.status, EffectStatus::Pending);
902    assert!(stored.approved, "the store must persist approval");
903
904    let other = insert(&store, 2).await;
905    let other = drive(
906        &store,
907        other,
908        None,
909        &[Transition::RequestApproval, Transition::Deny],
910        t(3),
911    )
912    .await;
913    assert_eq!(other.status, EffectStatus::Rejected);
914    assert!(!reload(&store, other.id).await.approved);
915}
916
917/// Runs effect `n` through `transitions` at time `at`, then releases its
918/// lease.
919async fn settle<S: EffectStore>(
920    store: &S,
921    n: u32,
922    transitions: &[Transition],
923    at: u64,
924) -> EffectId {
925    let record = insert(store, n).await;
926    let lease = store
927        .acquire_lease(record.id, &worker("a"), t(at), TTL)
928        .await
929        .unwrap();
930    drive(store, record, Some(&lease), transitions, t(at)).await;
931    store.release_lease(&lease).await.unwrap();
932    lease.effect_id
933}
934
935async fn pruning_removes_only_settled_idle_old_records<S: EffectStore>(store: S) {
936    let committed = [Transition::StartAttempt, Transition::Succeeded];
937    let old_a = settle(&store, 0, &committed, 10).await;
938    let failed = settle(
939        &store,
940        1,
941        &[Transition::StartAttempt, Transition::FailedDefinitively],
942        10,
943    )
944    .await;
945    let young = settle(&store, 2, &committed, 50).await;
946    let unknown = settle(
947        &store,
948        3,
949        &[Transition::StartAttempt, Transition::OutcomeUnknown],
950        10,
951    )
952    .await;
953    let escalated = settle(
954        &store,
955        6,
956        &[
957            Transition::StartAttempt,
958            Transition::OutcomeUnknown,
959            Transition::Escalate,
960        ],
961        10,
962    )
963    .await;
964    let leased = settle(&store, 4, &committed, 10).await;
965    let old_b = settle(&store, 5, &committed, 10).await;
966    // A compensation is about to start on `leased`: it holds a live lease.
967    store
968        .acquire_lease(leased, &worker("b"), t(95), TTL)
969        .await
970        .unwrap();
971
972    let now = t(100);
973    let minute = Duration::from_secs(60);
974    let prune = |status, older_than, limit| {
975        let store = &store;
976        async move {
977            store
978                .prune(PruneQuery::new(status, older_than, now).limit(limit))
979                .await
980                .unwrap()
981        }
982    };
983    let exists = |id| {
984        let store = &store;
985        async move { store.get(id).await.unwrap().is_some() }
986    };
987
988    assert_eq!(
989        prune(EffectStatus::Committed, minute, 1).await,
990        1,
991        "the limit caps a batch"
992    );
993    assert!(!exists(old_a).await, "lowest id first");
994    assert!(exists(old_b).await);
995    assert_eq!(
996        prune(EffectStatus::Committed, minute, 100).await,
997        1,
998        "only old_b is left to prune: young is too young, leased is leased"
999    );
1000    assert!(!exists(old_b).await);
1001    assert!(exists(young).await && exists(leased).await);
1002    assert_eq!(prune(EffectStatus::Committed, minute, 100).await, 0);
1003
1004    // Spelled out, not `is_settled`, so the suite does not share the
1005    // store's definition.
1006    let settled = [
1007        EffectStatus::Committed,
1008        EffectStatus::Failed,
1009        EffectStatus::Rejected,
1010        EffectStatus::Compensated,
1011    ];
1012    for status in ALL_STATUSES.into_iter().filter(|s| !settled.contains(s)) {
1013        assert_eq!(
1014            prune(status, Duration::ZERO, 100).await,
1015            0,
1016            "{status} is not settled and is never pruned"
1017        );
1018    }
1019    assert!(exists(unknown).await && exists(escalated).await);
1020    assert_eq!(
1021        prune(
1022            EffectStatus::Failed,
1023            Duration::from_secs(200_000_000_000),
1024            100
1025        )
1026        .await,
1027        0,
1028        "an age older than any record prunes nothing"
1029    );
1030    assert_eq!(prune(EffectStatus::Failed, minute, 100).await, 1);
1031    assert!(!exists(failed).await);
1032
1033    pruned_records_leave_nothing_behind(&store, old_a, young).await;
1034}
1035
1036/// After `pruned` (key 0) was pruned and `kept` was not.
1037async fn pruned_records_leave_nothing_behind<S: EffectStore>(
1038    store: &S,
1039    pruned: EffectId,
1040    kept: EffectId,
1041) {
1042    // The audit trail goes with the record; others keep theirs.
1043    assert!(matches!(
1044        store.events(pruned).await,
1045        Err(StoreError::NotFound(id)) if id == pruned
1046    ));
1047    assert_eq!(store.events(kept).await.unwrap().len(), 2);
1048    // The key is free again: the next request is a new effect.
1049    assert!(store.get_by_key(&key(0)).await.unwrap().is_none());
1050    let again = store.insert_or_get(new_effect(0)).await.unwrap();
1051    assert!(again.inserted);
1052    assert_ne!(again.record.id, pruned);
1053    assert_eq!(again.record.status, EffectStatus::Pending);
1054    assert_eq!(store.events(again.record.id).await.unwrap(), Vec::new());
1055}
agent_effects_store/testkit.rs

agent_effects_store/
testkit.rs
