1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
//! Durable delivery of one item to one host destination.
use async_trait::async_trait;
use time::OffsetDateTime;
use crate::error::DomainError;
use crate::value_objects::{
CeremonyId, DeliveryExpiryCause, DeliveryFailureReason, DurationMs, FollowReplacement,
HostAgentIncarnation, HostDeliveryId, HostDeliveryLease, HostDeliveryObservation,
HostDeliveryRecord, HostDeliveryTarget, IntegratorFence, ProcessedActionRef,
};
use super::{
AckOutcome, DeliveryFailureOutcome, EnqueueOutcome, HostActivationOutcome, HostDeliveryFilter,
HostDeliveryPage, HostDeliveryPageLimit, HostDeliveryQuery, LeasedDelivery, ProcessedOutcome,
RecordedActivation, SupersessionOutcome,
};
/// Persistence contract for work handed out to hosts.
///
/// Modelled on the named event cursor rather than on a queue: an
/// exclusive expiring lease, a counted attempt, and an end that stays
/// visible. A queue whose acknowledgement consumes the item cannot
/// answer the question an operator actually asks, which is what
/// happened to the question nobody ever came back about.
#[async_trait]
pub trait HostDeliveryLedgerPort: Send + Sync {
/// Offer a delivery, or recognise the one already held.
async fn enqueue(&self, record: HostDeliveryRecord) -> Result<EnqueueOutcome, DomainError>;
/// Hand out, exclusively and with an expiry, what a host can take.
///
/// A delivery whose lease has expired is offerable again: the lease
/// bounds the exclusion, so a host that died holding one strands
/// nothing.
async fn lease(
&self,
filter: &HostDeliveryFilter,
owner: &HostAgentIncarnation,
now: OffsetDateTime,
duration: DurationMs,
limit: HostDeliveryPageLimit,
) -> Result<Vec<LeasedDelivery>, DomainError>;
/// Record what the host said about a delivery it holds.
async fn acknowledge(
&self,
lease: &HostDeliveryLease,
observation: &HostDeliveryObservation,
now: OffsetDateTime,
) -> Result<AckOutcome, DomainError>;
/// Close a delivery the host has acted on.
///
/// Only from acknowledged, because acting on something never
/// received is not a thing that can have happened. The fence, when
/// the caller has one, is checked before anything is written.
async fn mark_processed(
&self,
delivery_id: &HostDeliveryId,
owner: &HostAgentIncarnation,
fence: Option<IntegratorFence>,
action: &ProcessedActionRef,
now: OffsetDateTime,
) -> Result<ProcessedOutcome, DomainError>;
/// Write down what an activation adapter did with an envelope.
///
/// Not the leased path, and deliberately so. An activation is a
/// push: the engine woke a host and holds nothing, so requiring a
/// lease here would mean taking one out against itself to record
/// that it made a phone call. Reaching a host is transport and
/// never processing, so a delivered record stays offerable and the
/// host still comes and takes the work under its own lease.
async fn record_activation(
&self,
delivery_id: &HostDeliveryId,
outcome: &HostActivationOutcome,
now: OffsetDateTime,
) -> Result<RecordedActivation, DomainError>;
/// Count a failed attempt, and retry or give up by the policy.
async fn mark_failed(
&self,
lease: &HostDeliveryLease,
reason: &DeliveryFailureReason,
now: OffsetDateTime,
) -> Result<DeliveryFailureOutcome, DomainError>;
/// Give a lease back without counting an attempt against it.
async fn release(
&self,
lease: &HostDeliveryLease,
now: OffsetDateTime,
) -> Result<(), DomainError>;
/// Expire what has run out: leases, and acknowledgements nobody closed.
async fn expire(&self, now: OffsetDateTime) -> Result<Vec<HostDeliveryId>, DomainError>;
/// Give up on one offer that is no longer worth making.
///
/// The sweeps below are about a reason that stopped applying to a
/// whole ceremony; this is about one item, and the only caller
/// that has one is the thing keeping a queue bounded. It stays in
/// the ledger with its cause, because a host that was never told
/// something needs somebody to be able to read why.
async fn abandon(
&self,
delivery_id: &HostDeliveryId,
cause: DeliveryExpiryCause,
now: OffsetDateTime,
) -> Result<Option<HostDeliveryRecord>, DomainError>;
/// Give up on everything still open for one ceremony, naming why.
///
/// The sweep above is about time running out; this is about the
/// reason for asking running out — a ceremony that ended, a
/// binding that was revoked. The offers stay in the ledger with
/// the cause on them, because "nobody was ever asked this" is
/// exactly what an operator needs to be able to read afterwards,
/// and a row that quietly stayed queued forever says the opposite.
/// Terminal records are untouched.
async fn expire_ceremony(
&self,
ceremony_id: &CeremonyId,
cause: DeliveryExpiryCause,
now: OffsetDateTime,
) -> Result<Vec<HostDeliveryId>, DomainError>;
/// Follow a destination that was replaced, or leave its work behind.
async fn supersede(
&self,
previous: &HostDeliveryTarget,
replacement: &HostDeliveryTarget,
follow: FollowReplacement,
now: OffsetDateTime,
) -> Result<SupersessionOutcome, DomainError>;
/// One delivery, whatever state it is in.
async fn get(&self, id: &HostDeliveryId) -> Result<Option<HostDeliveryRecord>, DomainError>;
/// One page of what the ledger holds.
async fn list(&self, query: &HostDeliveryQuery) -> Result<HostDeliveryPage, DomainError>;
}