alice.pds.demo.boltless.dev
This repository has no description
11 kB
329 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1//! # How we go about receiving a push!
2//!
3//! `land` will run the pack ingest and the pull-link lookup at the same time,
4//! then will merge both into a single post-receive pass.
5//! Each ref-update claims its event cursor during ingest
6//! and fills the payload in afterward,
7//! such that events replay in ref-order even though post-receive was what computed them.
8//!
9//! Or should I say:
10//!
11//! ```text
12//! request task blocking pool
13//! --------------------------------- ------------------------------
14//! read the push preamble
15//! |
16//! +---- spawn -------------------> open repo
17//! | |
18//! creates a branch? no -> no link receive pack
19//! | yes |
20//! look up pull link seal: 1 cursor per ref update
21//! | repo-DID -> owner, rkey, handle |
22//! v v
23//! join <---------------------------- updates paired w/ reservations
24//! |
25//! any refs applied? no -> no messages
26//! | yes
27//! owner & rkey for this repo-DID
28//! |
29//! ci flags from push options
30//! |
31//! +---- spawn -------------------> ack lines
32//! | |
33//! | post_receive fulfills each
34//! | | reservation
35//! v v
36//! messages <------------------------ ..and returns its messages
37//! |
38//! note this push for maintenance
39//! |
40//! frame report for the client
41//! ```
42//!
43//! *Figure 1: the tasks of a push.*
44//!
45//! `EventLog::replay` finishes at the oldest pending cursor,
46//! meaning that a single reservation will hide every later-event until it resolves.
47//! Every exit from `land` therefore has to either fulfill each one or drop it,
48//! and `Reservation`'s `Drop` clears
49//! the pending cursor such that replay advances again.
50//!
51//! ```text
52//! one Reservation
53//! ---------------
54//! reserve -> pending, replay finishes here
55//! |
56//! +-- post_receive fulfills it -----> event is visible
57//! +-- receive errors after seal ----> dropped, logged
58//! +-- receive task panics ----------> dropped, silent
59//! +-- post-receive task panics -----> dropped, logged
60//! +-- caller drops the land future -> dropped, silent
61//! ```
62//!
63//! *Figure 2: every way in which a single reservation can end.*
64//!
65//! Note that last path in Figure 2 leaves the receive task running
66//! with nowhere to return its result,
67//! so the reservations inside it drop alongside the discarded value.
68
69use std::cell::RefCell;
70use std::sync::Arc;
71
72use knot_atproto::Atproto;
73use knot_cob::CobHome;
74use knot_events::{EventLog, Reservation};
75use knot_git::{Layout, RefUpdate, Repo};
76use knot_index::{Index, Resolved};
77use knot_maintenance::{MaintenanceHandle, PushBytes};
78use knot_messages::{Catalog, PushAckKey, count_refs};
79use knot_pack::{
80 PackError, PackLimits, PushGuard, ReceiveOutcome, ReceivedPack, frame_report,
81 receive_pack_guarded_streamed, receive_preflight,
82};
83use knot_postreceive::{Actor, Ci, LanguagesPushBudget, OwnerLabel, PullLink, post_receive};
84use knot_resource::ResolveSlots;
85use knot_runtime::{Clock, HttpTransport};
86use knot_types::{
87 AccountDid, ActorId, AppviewEndpoint, CiLogsAddr, Handle, KnotHostname, OwnerDid, PushOptions,
88 RepoDid,
89};
90
91type Applied = Vec<(RefUpdate, Reservation)>;
92
93pub struct Push<'a, H: HttpTransport, C: Clock> {
94 pub layout: &'a Layout,
95 pub repo_did: &'a RepoDid,
96 pub received: ReceivedPack,
97 pub limits: PackLimits,
98 pub knot_actor: ActorId,
99 pub committer: AccountDid,
100 pub events: Arc<EventLog<C>>,
101 pub index: &'a Index,
102 pub atproto: &'a Atproto<H, C>,
103 pub resolve_slots: &'a ResolveSlots,
104 pub appview: &'a AppviewEndpoint,
105 pub maintenance: &'a MaintenanceHandle,
106 pub hostname: &'a KnotHostname,
107 pub languages_push_budget: LanguagesPushBudget,
108 pub ci_logs: Option<CiLogsAddr>,
109 pub catalog: Arc<Catalog>,
110}
111
112pub async fn land<H: HttpTransport, C: Clock>(push: Push<'_, H, C>) -> Result<Vec<u8>, PackError> {
113 let Push {
114 layout,
115 repo_did,
116 received,
117 limits,
118 knot_actor,
119 committer,
120 events,
121 index,
122 atproto,
123 resolve_slots,
124 appview,
125 maintenance,
126 hostname,
127 languages_push_budget,
128 catalog,
129 ci_logs,
130 } = push;
131
132 let preflight = receive_preflight(received.preamble());
133 let body_len = received.len();
134 let home = CobHome::from(repo_did);
135
136 let receive = {
137 let layout = layout.clone();
138 let did = repo_did.clone();
139 let events = Arc::clone(&events);
140 let catalog = Arc::clone(&catalog);
141 tokio::task::spawn_blocking(
142 move || -> Result<(Repo, ReceiveOutcome, Applied), PackError> {
143 let repo = layout.open(&did)?;
144 let guard = PushGuard {
145 cob_authority: knot_actor,
146 home,
147 messages: Arc::clone(&catalog),
148 };
149 let stash: RefCell<Applied> = RefCell::new(Vec::new());
150 let seal = |updates: &[RefUpdate]| {
151 updates.iter().for_each(|update| {
152 stash.borrow_mut().push((update.clone(), events.reserve()))
153 });
154 };
155 let outcome = match receive_pack_guarded_streamed(
156 &repo,
157 &received,
158 &limits,
159 &guard,
160 &seal,
161 &catalog.reject,
162 ) {
163 Ok(outcome) => outcome,
164 Err(error) => {
165 match stash.borrow().len() {
166 0 => {}
167 sealed => tracing::error!(
168 repo = did.as_str(),
169 sealed,
170 "receive failure dropped the events for the sealed ref updates"
171 ),
172 }
173 return Err(error);
174 }
175 };
176 Ok((repo, outcome, stash.into_inner()))
177 },
178 )
179 };
180 let pull = async {
181 match preflight.creates_branch {
182 true => resolve_pull_link(index, atproto, resolve_slots, appview, repo_did).await,
183 false => None,
184 }
185 };
186 let (received, pull) = tokio::join!(receive, pull);
187 let (repo, outcome, applied) = match received {
188 Ok(Ok(triple)) => triple,
189 Ok(Err(error)) => return Err(error),
190 Err(_) => return Err(PackError::Pack("receive-pack task panicked".to_string())),
191 };
192
193 let messages = match applied.is_empty() {
194 true => Vec::new(),
195 false => {
196 let owner = registry_owner(index, repo_did);
197 let ci = ci_from_push_options(&outcome.push_options, ci_logs.clone());
198 let push_options = outcome.push_options.clone();
199 let did = repo_did.clone();
200 let catalog = Arc::clone(&catalog);
201 let knot = hostname.clone();
202 tokio::task::spawn_blocking(move || {
203 let actor = Actor {
204 committer,
205 owner,
206 repo: did,
207 };
208 let ack = catalog.push.ack.lines(|key| match key {
209 PushAckKey::Knot => knot.as_str().to_string(),
210 PushAckKey::Refs => count_refs(applied.len()),
211 });
212 ack.into_iter()
213 .chain(post_receive(
214 &repo,
215 &actor,
216 applied,
217 &ci,
218 &push_options,
219 pull.as_ref(),
220 languages_push_budget,
221 &catalog.push,
222 ))
223 .collect()
224 })
225 .await
226 .unwrap_or_else(|error| {
227 tracing::error!(repo = repo_did.as_str(), %error, "post-receive task panicked and dropped the ref-update events for this push");
228 Vec::new()
229 })
230 }
231 };
232 maintenance.note_push(repo_did, PushBytes::new(body_len as u64));
233 Ok(frame_report(&outcome.report, &messages, outcome.side_band))
234}
235
236fn registry_owner(index: &Index, repo: &RepoDid) -> Option<OwnerDid> {
237 let ready = |owner: Resolved<Option<OwnerDid>>| match owner {
238 Resolved::Ready(owner) => owner,
239 Resolved::Warming => None,
240 };
241 match index.owner_of(repo) {
242 resolved @ Resolved::Ready(_) => ready(resolved),
243 Resolved::Warming => {
244 if let Err(error) = index.refresh_registry() {
245 tracing::warn!(
246 repo = repo.as_str(),
247 %error,
248 "registry refresh during post-receive failed, ref-update event omits the owner"
249 );
250 }
251 ready(index.owner_of(repo))
252 }
253 }
254}
255
256#[derive(Debug, Clone, Copy, PartialEq, Eq)]
257enum PushDirective {
258 SkipCi,
259 VerboseCi,
260}
261
262impl PushDirective {
263 fn parse(option: &str) -> Option<Self> {
264 match option {
265 "skip-ci" | "ci-skip" => Some(Self::SkipCi),
266 "verbose-ci" | "ci-verbose" => Some(Self::VerboseCi),
267 _ => None,
268 }
269 }
270}
271
272pub fn ci_from_push_options(options: &PushOptions, logs: Option<CiLogsAddr>) -> Ci {
273 let directives: Vec<PushDirective> = options
274 .as_slice()
275 .iter()
276 .filter_map(|option| PushDirective::parse(option.as_str()))
277 .collect();
278 match directives.contains(&PushDirective::SkipCi) {
279 true => Ci::Skip,
280 false => Ci::Compile {
281 logs,
282 verbose: directives.contains(&PushDirective::VerboseCi),
283 },
284 }
285}
286
287async fn resolve_pull_link<H: HttpTransport, C: Clock>(
288 index: &Index,
289 atproto: &Atproto<H, C>,
290 resolve_slots: &ResolveSlots,
291 appview: &AppviewEndpoint,
292 repo_did: &RepoDid,
293) -> Option<PullLink> {
294 let owner = match index.owner_of(repo_did) {
295 Resolved::Ready(Some(owner)) => owner,
296 _ => return None,
297 };
298 let rkey = match index.rkey_of(repo_did) {
299 Resolved::Ready(Some(rkey)) => rkey,
300 _ => return None,
301 };
302 Some(PullLink {
303 appview: appview.clone(),
304 owner: resolve_owner_label(atproto, resolve_slots, &owner).await,
305 rkey,
306 })
307}
308
309async fn resolve_owner_label<H: HttpTransport, C: Clock>(
310 atproto: &Atproto<H, C>,
311 resolve_slots: &ResolveSlots,
312 owner: &OwnerDid,
313) -> OwnerLabel {
314 let did = AccountDid::from(owner.clone());
315 match resolve_handle(atproto, resolve_slots, &did).await {
316 Some(handle) => OwnerLabel::Handle(handle),
317 None => OwnerLabel::Did(owner.clone()),
318 }
319}
320
321pub async fn resolve_handle<H: HttpTransport, C: Clock>(
322 atproto: &Atproto<H, C>,
323 resolve_slots: &ResolveSlots,
324 did: &AccountDid,
325) -> Option<Handle> {
326 let _permit = resolve_slots.try_acquire()?;
327 let identity = atproto.resolve_identity(did).await.ok()?;
328 identity.primary_handle().cloned()
329}