From 43d986187de35090b4aa665fc8b381409919ef9f Mon Sep 17 00:00:00 2001 From: dawn Date: Mon, 26 Oct 2026 20:05:57 +0300 Subject: [PATCH] knot2/{receive,xrpc}: print what the spindle says about a push instead of a knot ci logs address Signed-off-by: dawn --- docs/src/content/docs/knots/index.md | 3 +- .../src/content/docs/reference/knot-config.md | 6 - docs/src/content/docs/self-hosting/knot.md | 9 +- knot2/crates/knot-atproto/src/lib.rs | 17 ++- knot2/crates/knot-config/src/lib.rs | 11 -- knot2/crates/knot-messages/src/lib.rs | 5 - knot2/crates/knot-postreceive/src/lib.rs | 31 ++-- .../knot-postreceive/tests/post_receive.rs | 111 +-------------- knot2/crates/knot-receive/src/lib.rs | 134 +++++++++++++++--- knot2/crates/knot-server/src/main.rs | 12 +- knot2/crates/knot-sim/src/harness.rs | 1 - knot2/crates/knot-sim/tests/lfs_roundtrip.rs | 2 - knot2/crates/knot-sim/tests/ssh.rs | 1 - .../knot-ssh/examples/ephemeral_knot.rs | 1 - knot2/crates/knot-ssh/src/exec.rs | 1 - knot2/crates/knot-ssh/src/lib.rs | 6 +- knot2/crates/knot-ssh/tests/ssh_push.rs | 3 - knot2/crates/knot-types/src/ids.rs | 91 ------------ knot2/crates/knot-types/src/lib.rs | 4 +- knot2/crates/knot-workflow/src/lib.rs | 79 ++--------- knot2/crates/knot-xrpc/src/lib.rs | 3 +- knot2/crates/knot-xrpc/src/outbox.rs | 84 +++++++++-- knot2/crates/knot-xrpc/src/receive.rs | 1 - knot2/crates/knot-xrpc/src/refrecords.rs | 13 +- knot2/crates/knot-xrpc/src/tests.rs | 1 - knot2/crates/knot-xrpc/tests/common/mod.rs | 1 - knot2/example.toml | 8 -- 27 files changed, 245 insertions(+), 394 deletions(-) diff --git a/docs/src/content/docs/knots/index.md b/docs/src/content/docs/knots/index.md index b0c775a6c..d4efcc9fa 100644 --- a/docs/src/content/docs/knots/index.md +++ b/docs/src/content/docs/knots/index.md @@ -108,7 +108,8 @@ usually from a git credential helper. After a push, git prints **push messages**: an acknowledgement, an "Open pull request" link for new branches, workflow compile diagnostics, and a -CI-logs SSH hint when `[ci] logs_addr` is set. The text comes from +whatever the repository's spindle answered about the pipeline it scheduled, +such as how to follow its logs over SSH. The text comes from `[messages.*]` templates. Pass `-o skip-ci` to skip CI for a push. A Knot won't allow any pushes to the following reserved refs: `refs/cobs/*`, diff --git a/docs/src/content/docs/reference/knot-config.md b/docs/src/content/docs/reference/knot-config.md index 4dd6b0143..e04e69ec1 100644 --- a/docs/src/content/docs/reference/knot-config.md +++ b/docs/src/content/docs/reference/knot-config.md @@ -236,12 +236,6 @@ hosts. Git data stays readable. | `enabled` | `KNOT_HOMEPAGE_ENABLED` | `true` | Serve an HTML homepage at `/`. | | `path` | `KNOT_HOMEPAGE_PATH` | unset | Custom HTML file for the homepage. Unset serves the built-in page. | -## `[ci]` - -| Key | Env var | Default | Description | -| ----------- | ------------------- | ------- | ---------------------------------------------------------------------------------------------------------- | -| `logs_addr` | `KNOT_CI_LOGS_ADDR` | unset | `host:port` of an SSH endpoint for browsing CI logs. When set, push output includes an `ssh -t -p …` hint. | - ## `[messages]` The `[messages.push]`, `[messages.fetch]`, `[messages.reject]`, diff --git a/docs/src/content/docs/self-hosting/knot.md b/docs/src/content/docs/self-hosting/knot.md index 56d7072be..1d2fca05e 100644 --- a/docs/src/content/docs/self-hosting/knot.md +++ b/docs/src/content/docs/self-hosting/knot.md @@ -379,9 +379,12 @@ once they are older than `lfs.gc_grace_secs` (two weeks). ## CI logs over SSH -With `[ci] logs_addr` set, push output ends with -`ssh -t -p `. The `cilogs` service -answers that address (`nix build .#cilogs`, or `nixosModules.cilogs`). +When a push schedules a pipeline, the knot prints what the spindle +answered to its notification, for example +`ssh -t -p ` from a spindle with +`SPINDLE_SERVER_LOGS_ADDR` set. The knot waits at most two seconds for +that answer. The `cilogs` service answers that address +(`nix build .#cilogs`, or `nixosModules.cilogs`). It keeps no state besides its host key: per connection it reads the repository's manifest from its knot, takes the spindle from the manifest's `did:web`, and follows the run over that spindle's diff --git a/knot2/crates/knot-atproto/src/lib.rs b/knot2/crates/knot-atproto/src/lib.rs index 6187d15eb..c1086a5a6 100644 --- a/knot2/crates/knot-atproto/src/lib.rs +++ b/knot2/crates/knot-atproto/src/lib.rs @@ -186,8 +186,11 @@ impl AtprotoError { /// What a spindle said back to `notifyEvent`. #[derive(Debug, Clone, PartialEq, Eq)] pub enum SpindleAnswer { - /// The spindle-local id of the pipeline it scheduled. - Scheduled(String), + /// The spindle-local id of the pipeline it scheduled, and any text it has for the pusher. + Scheduled { + pipeline: String, + message: Option, + }, Nothing, Malformed(String), } @@ -811,14 +814,18 @@ impl Atproto { #[derive(Deserialize)] struct NotifyOutput { pipeline: Option, + message: Option, } - let output: NotifyOutput = match serde_json::from_slice(&response.body) { + let NotifyOutput { pipeline, message } = match serde_json::from_slice(&response.body) { Ok(output) => output, Err(error) => return Ok(SpindleAnswer::Malformed(error.to_string())), }; - Ok(match output.pipeline { + Ok(match pipeline { None => SpindleAnswer::Nothing, - Some(id) if is_pipeline_id(&id) => SpindleAnswer::Scheduled(id), + Some(id) if is_pipeline_id(&id) => SpindleAnswer::Scheduled { + pipeline: id, + message, + }, Some(id) => SpindleAnswer::Malformed(format!("pipeline id {id:?} isn't usable")), }) } diff --git a/knot2/crates/knot-config/src/lib.rs b/knot2/crates/knot-config/src/lib.rs index 552007766..26a6fbf26 100644 --- a/knot2/crates/knot-config/src/lib.rs +++ b/knot2/crates/knot-config/src/lib.rs @@ -50,8 +50,6 @@ pub struct KnotConfig { #[config(nested)] pub homepage: HomepageConfig, - #[config(nested)] - pub ci: CiConfig, #[config(nested)] pub messages: knot_messages::MessagesConfig, } @@ -181,12 +179,6 @@ pub struct RepoConfig { pub write_refill_ms: u64, } -#[derive(Debug, Config)] -pub struct CiConfig { - #[config(env = "KNOT_CI_LOGS_ADDR")] - pub logs_addr: Option, -} - #[derive(Debug, Config)] pub struct HomepageConfig { #[config(env = "KNOT_HOMEPAGE_ENABLED", default = true)] @@ -1415,9 +1407,6 @@ mod tests { enabled: true, path: None, }, - ci: CiConfig { - logs_addr: Some("logs.oyster.cafe:3333".to_string()), - }, } } diff --git a/knot2/crates/knot-messages/src/lib.rs b/knot2/crates/knot-messages/src/lib.rs index f82ef5971..220c3d60b 100644 --- a/knot2/crates/knot-messages/src/lib.rs +++ b/knot2/crates/knot-messages/src/lib.rs @@ -126,7 +126,6 @@ keys! { KnotKey { Knot = "knot" } PushAckKey { Knot = "knot", Refs = "refs" } UrlKey { Url = "url" } - CiLogsKey { Host = "host", Port = "port", Repo = "repo", Sha = "sha" } GreetingKey { User = "user", Knot = "knot" } AuthorizedKey { Authorized = "authorized" } CountKey { Count = "count" } @@ -156,10 +155,6 @@ message_group! { ], pipeline_clean: Lines @ "KNOT_MESSAGES_PUSH_PIPELINE_CLEAN" = ["pipeline compiled with no diagnostics"], pipeline_none: Lines @ "KNOT_MESSAGES_PUSH_PIPELINE_NONE" = ["no pipelines to compile"], - ci_logs: Lines @ "KNOT_MESSAGES_PUSH_CI_LOGS" = [ - "-> Browse CI logs in your terminal:", - " ssh -t -p {port} {host} {repo} {sha}" - ], } } diff --git a/knot2/crates/knot-postreceive/src/lib.rs b/knot2/crates/knot-postreceive/src/lib.rs index 1f54a16b0..c3b327853 100644 --- a/knot2/crates/knot-postreceive/src/lib.rs +++ b/knot2/crates/knot-postreceive/src/lib.rs @@ -1,8 +1,8 @@ use knot_git::{EntryKind, RefUpdate, Repo}; -use knot_messages::{CiLogsKey, PushMessages, UrlKey}; +use knot_messages::{PushMessages, UrlKey}; use knot_types::{ - AccountDid, AppviewEndpoint, BranchName, ChangedFiles, CiLogsAddr, Handle, Listing, Oid, - OriginUrl, OwnerDid, RefName, RefTransition, RepoDid, RepoPath, RepoRkey, + AccountDid, AppviewEndpoint, BranchName, ChangedFiles, Handle, Listing, Oid, OriginUrl, + OwnerDid, RefName, RefTransition, RepoDid, RepoPath, RepoRkey, }; use knot_workflow::{Compiled, RawWorkflow, Trigger, WorkflowName}; use url::Url; @@ -16,10 +16,7 @@ pub struct Actor { pub enum Ci { Skip, - Compile { - logs: Option, - verbose: bool, - }, + Compile { verbose: bool }, } pub enum OwnerLabel { @@ -230,7 +227,7 @@ fn ci_messages( changed: &ChangedFiles, context: &PushContext, ) -> Vec { - let Ci::Compile { logs, verbose } = context.ci else { + let Ci::Compile { verbose } = context.ci else { return Vec::new(); }; let Some(new) = transition.new_oid() else { @@ -238,18 +235,16 @@ fn ci_messages( }; let templates = context.messages; let raws = read_workflows(repo, new); - let compiled = knot_workflow::compile( + let Compiled { + workflows, + diagnostics, + } = knot_workflow::compile( &raws, &Trigger::Push { ref_name: name.clone(), }, changed, ); - let listed = compiled.any_listed_match(); - let Compiled { - workflows, - diagnostics, - } = compiled; let mut messages = diagnostics.errors; if *verbose { let clean = messages.is_empty() && diagnostics.warnings.is_empty(); @@ -260,14 +255,6 @@ fn ci_messages( (false, false) => {} } } - if let Some(addr) = logs.as_ref().filter(|_| listed) { - messages.extend(templates.ci_logs.lines(|key| match key { - CiLogsKey::Host => addr.host().to_string(), - CiLogsKey::Port => addr.port().to_string(), - CiLogsKey::Repo => context.actor.repo.to_string(), - CiLogsKey::Sha => new.to_hex(), - })); - } messages } diff --git a/knot2/crates/knot-postreceive/tests/post_receive.rs b/knot2/crates/knot-postreceive/tests/post_receive.rs index 98147ce7e..a100738ee 100644 --- a/knot2/crates/knot-postreceive/tests/post_receive.rs +++ b/knot2/crates/knot-postreceive/tests/post_receive.rs @@ -3,8 +3,7 @@ use std::path::Path; use knot_git::{Layout, RefUpdate, Repo}; use knot_postreceive::{Actor, Ci, OwnerLabel, PullLink, post_receive}; use knot_types::{ - AccountDid, AppviewEndpoint, BranchName, CiLogsAddr, Handle, Oid, OriginUrl, RefName, RepoDid, - RepoRkey, + AccountDid, AppviewEndpoint, BranchName, Handle, Oid, OriginUrl, RefName, RepoDid, RepoRkey, }; const DID: &str = "did:plc:limpet"; @@ -92,17 +91,7 @@ fn pull() -> PullLink { } fn compile(verbose: bool) -> Ci { - Ci::Compile { - logs: None, - verbose, - } -} - -fn compile_with_logs(addr: &str) -> Ci { - Ci::Compile { - logs: Some(CiLogsAddr::new(addr).unwrap()), - verbose: false, - } + Ci::Compile { verbose } } fn run(world: &World, applied: &[RefUpdate], ci: &Ci, pull: Option<&PullLink>) -> Vec { @@ -140,48 +129,9 @@ fn create_feature(world: &World) -> Oid { } #[test] -fn a_configured_logs_address_yields_an_ssh_command_for_compiled_workflows() { - let world = world(); - commit_file( - &world.work, - ".tangled/workflows/ci.yml", - "engine: nixery.dev/x\nwhen:\n - event: push\n branch: ['**']\n", - "add ci", - ); - push(&world, "main"); - let head = oid(&world, "HEAD"); - - let applied = created("main", head); - let messages = run( - &world, - &applied, - &compile_with_logs("logs.oyster.cafe:3333"), - None, - ); - - assert!( - messages - .iter() - .any(|line| line.contains(&format!("ssh -t -p 3333 logs.oyster.cafe {DID} {head}"))), - "{messages:?}" - ); -} - -#[test] -fn a_push_without_workflows_yields_no_ssh_command_and_verbose_says_so() { +fn a_push_without_workflows_says_so_when_verbose() { let world = world(); let applied = create(&world, "main"); - let messages = run( - &world, - &applied, - &compile_with_logs("logs.oyster.cafe:3333"), - None, - ); - assert!( - messages.iter().all(|line| !line.contains("ssh -t")), - "{messages:?}" - ); - let messages = run(&world, &applied, &compile(true), None); assert!( messages @@ -225,61 +175,6 @@ fn verbose_keeps_the_warning_that_explains_why_nothing_compiled() { ); } -fn ci_yaml(paths: &str) -> String { - format!( - "engine: nixery.dev/x\nwhen:\n - event: push\n branch: ['**']\n paths: ['{paths}']\n" - ) -} - -#[test] -fn a_push_wider_than_the_record_prints_the_ssh_command_only_on_a_listed_glob_hit() { - let world = world(); - commit_file( - &world.work, - ".tangled/workflows/ci.yml", - &ci_yaml("never/**"), - "add ci", - ); - (0..20_000).for_each(|index| { - std::fs::write(world.work.join(format!("f{index}.txt")), "x\n").unwrap(); - }); - git(&world.work, &["add", "-A"]); - git(&world.work, &["commit", "-q", "-m", "many"]); - push(&world, "main"); - let unmatched = oid(&world, "HEAD"); - commit_file( - &world.work, - ".tangled/workflows/ci.yml", - &ci_yaml("f1.txt"), - "aim ci", - ); - push(&world, "main"); - let matched = oid(&world, "HEAD"); - - let messages = run( - &world, - &created("main", unmatched), - &compile_with_logs("logs.oyster.cafe:3333"), - None, - ); - assert!( - messages.iter().all(|line| !line.contains("ssh -t")), - "spindle reads the same truncated listing, rules the globs out, \ - and skips this run: {messages:?}" - ); - - let messages = run( - &world, - &created("main", matched), - &compile_with_logs("logs.oyster.cafe:3333"), - None, - ); - assert!( - messages.iter().any(|line| line.contains("ssh -t -p 3333")), - "spindle sees the same listed path and runs this workflow: {messages:?}" - ); -} - #[test] fn a_new_non_default_branch_yields_a_pull_request_link() { let world = world(); diff --git a/knot2/crates/knot-receive/src/lib.rs b/knot2/crates/knot-receive/src/lib.rs index cdc7c85f7..8b93f5d38 100644 --- a/knot2/crates/knot-receive/src/lib.rs +++ b/knot2/crates/knot-receive/src/lib.rs @@ -1,7 +1,9 @@ //! # How we go about receiving a push! //! //! `land` will run the pack ingest and the pull-link lookup at the same time, -//! then will merge both into a single post-receive pass. +//! then will merge both into a single post-receive pass. Whatever the repo's +//! spindle says about the push comes last, and we only wait +//! `SPINDLE_REPLY_WAIT` for it. //! The pack receiver seals each applied ref-update while the ref lock is held, //! so post-receive sees exactly the updates that moved a ref, in ref order. //! @@ -30,6 +32,8 @@ //! | post_receive runs per update, //! v then the ref projection //! messages <------------------------ ..and returns its messages +//! | and the spindle's reply slots +//! wait for the spindle's replies //! | //! note this push for maintenance //! | @@ -40,6 +44,7 @@ use std::cell::RefCell; use std::sync::Arc; +use std::time::Duration; use knot_atproto::Atproto; use knot_cob::CobHome; @@ -55,12 +60,18 @@ use knot_postreceive::{Actor, Ci, OwnerLabel, PullLink, post_receive}; use knot_resource::ResolveSlots; use knot_runtime::{Clock, HttpTransport}; use knot_types::{ - AccountDid, ActorId, AppviewEndpoint, CiLogsAddr, Handle, KnotHostname, OwnerDid, PushOptions, - RepoDid, + AccountDid, ActorId, AppviewEndpoint, Handle, KnotHostname, OwnerDid, PushOptions, RepoDid, }; +use tokio::sync::oneshot; type Applied = Vec; +pub type SpindleReply = oneshot::Receiver>; + +const SPINDLE_REPLY_WAIT: Duration = Duration::from_secs(2); +const MAX_SPINDLE_LINES: usize = 10; +const MAX_SPINDLE_CHARS: usize = 1000; + pub trait RefSurface: Send + Sync { fn project( &self, @@ -69,7 +80,7 @@ pub trait RefSurface: Send + Sync { applied: &[RefUpdate], pusher: &AccountDid, options: &PushOptions, - ); + ) -> Vec; } pub struct Push<'a, H: HttpTransport, C: Clock> { @@ -85,7 +96,6 @@ pub struct Push<'a, H: HttpTransport, C: Clock> { pub appview: &'a AppviewEndpoint, pub maintenance: &'a MaintenanceHandle, pub hostname: &'a KnotHostname, - pub ci_logs: Option, pub catalog: Arc, pub ref_surface: Option>, } @@ -105,7 +115,6 @@ pub async fn land(push: Push<'_, H, C>) -> Result(push: Push<'_, H, C>) -> Result return Err(PackError::Pack("receive-pack task panicked".to_string())), }; - let messages = match applied.is_empty() { - true => Vec::new(), + let (mut messages, replies) = match applied.is_empty() { + true => (Vec::new(), Vec::new()), false => { - let ci = ci_from_push_options(&outcome.push_options, ci_logs.clone()); + let ci = ci_from_push_options(&outcome.push_options); let push_options = outcome.push_options.clone(); let did = repo_did.clone(); let catalog = Arc::clone(&catalog); @@ -188,18 +197,22 @@ pub async fn land(push: Push<'_, H, C>) -> Result>() + let replies = match ref_surface.as_deref() { + Some(project) => { + project.project(&repo, &did, &applied, &committer, &push_options) + } + None => Vec::new(), + }; + (messages.collect::>(), replies) }) } .await .unwrap_or_else(|error| { tracing::error!(repo = repo_did.as_str(), %error, "post-receive task panicked"); - Vec::new() + (Vec::new(), Vec::new()) }), }; + messages.extend(spindle_messages(replies).await); maintenance.note_push(repo_did, PushBytes::new(body_len as u64)); Ok(frame_report(&outcome.report, &messages, outcome.side_band)) } @@ -220,7 +233,7 @@ impl PushDirective { } } -pub fn ci_from_push_options(options: &PushOptions, logs: Option) -> Ci { +pub fn ci_from_push_options(options: &PushOptions) -> Ci { let directives: Vec = options .as_slice() .iter() @@ -229,12 +242,58 @@ pub fn ci_from_push_options(options: &PushOptions, logs: Option) -> match directives.contains(&PushDirective::SkipCi) { true => Ci::Skip, false => Ci::Compile { - logs, verbose: directives.contains(&PushDirective::VerboseCi), }, } } +/// a spindle can send anything, keep only plain printable lines before it reaches a terminal +async fn spindle_messages(replies: Vec) -> Vec { + let deadline = tokio::time::Instant::now() + SPINDLE_REPLY_WAIT; + let mut said = Vec::new(); + for reply in replies { + if let Ok(Ok(Some(text))) = tokio::time::timeout_at(deadline, reply).await + && !said.contains(&text) + { + said.push(text); + } + } + said.iter().flat_map(|text| terminal_lines(text)).collect() +} + +fn terminal_lines(text: &str) -> Vec { + let mut plain = String::new(); + let mut chars = text.chars().take(MAX_SPINDLE_CHARS).peekable(); + while let Some(c) = chars.next() { + match c { + '\u{1b}' => skip_escape(&mut chars), + '\n' => plain.push(c), + c if c.is_control() => {} + c => plain.push(c), + } + } + plain + .lines() + .take(MAX_SPINDLE_LINES) + .map(str::to_string) + .collect() +} + +// csi runs to a final byte in @..~, osc to bel or esc \, anything else is one char +fn skip_escape(chars: &mut std::iter::Peekable>) { + match chars.next() { + Some('[') => { + chars.find(|c| ('@'..='~').contains(c)); + } + Some(']') => { + if chars.find(|c| matches!(c, '\u{7}' | '\u{1b}')) == Some('\u{1b}') { + chars.next_if_eq(&'\\'); + } + } + _ => {} + } +} + async fn resolve_pull_link( index: &Index, atproto: &Atproto, @@ -278,3 +337,46 @@ pub async fn resolve_handle( let identity = atproto.resolve_identity(did).await.ok()?; identity.primary_handle().cloned() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_spindles_text_loses_escape_sequences_and_control_characters_but_keeps_its_lines() { + let text = "-> \u{1b}[31;1mBrowse\u{1b}[0m CI logs:\r\n ssh -t \u{1b}]8;;https://evil.test\u{7}here\u{1b}]8;;\u{1b}\\\u{7}\u{8}\u{9b}2J"; + assert_eq!( + terminal_lines(text), + ["-> Browse CI logs:", " ssh -t here2J"] + ); + } + + #[test] + fn a_spindles_text_is_capped_in_lines_and_length() { + let many = (0..50).map(|n| format!("line {n}\n")).collect::(); + assert_eq!(terminal_lines(&many).len(), MAX_SPINDLE_LINES); + let long = "x".repeat(5 * MAX_SPINDLE_CHARS); + assert_eq!(terminal_lines(&long), ["x".repeat(MAX_SPINDLE_CHARS)]); + } + + #[tokio::test] + async fn each_distinct_spindle_message_is_printed_once_and_a_silent_spindle_is_not_waited_on() { + let reply = |text: Option<&str>| { + let (pusher, reply) = oneshot::channel(); + pusher.send(text.map(str::to_string)).unwrap(); + reply + }; + let (_silent, hung) = oneshot::channel(); + let started = tokio::time::Instant::now(); + let lines = spindle_messages(vec![ + reply(Some("follow it")), + reply(None), + reply(Some("follow it")), + hung, + reply(Some("and this")), + ]) + .await; + assert_eq!(lines, ["follow it", "and this"]); + assert!(started.elapsed() < SPINDLE_REPLY_WAIT + Duration::from_secs(1)); + } +} diff --git a/knot2/crates/knot-server/src/main.rs b/knot2/crates/knot-server/src/main.rs index b44a43a60..b4d98106a 100644 --- a/knot2/crates/knot-server/src/main.rs +++ b/knot2/crates/knot-server/src/main.rs @@ -28,7 +28,7 @@ use knot_config::HomepageSource; use knot_index::Index; use knot_runtime::{Clock, HttpTransport, OsEntropy, ReqwestHttp, SystemClock}; use knot_secrets::{MasterKey, SealedStore}; -use knot_types::{ActorId, BranchName, CiLogsAddr, ObjectCount}; +use knot_types::{ActorId, BranchName, ObjectCount}; use knot_xrpc::XrpcState; use tower_http::services::ServeFile; @@ -555,14 +555,6 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { ) as usize), }; - let ci_logs = config - .ci - .logs_addr - .as_deref() - .map(CiLogsAddr::new) - .transpose() - .context("ci.logs_addr must be host:port")?; - let homepage = config.homepage.source(); let catalog = Arc::new( knot_messages::Catalog::parse(&config.messages).context("parse message templates")?, @@ -611,7 +603,6 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { admission, max_pack_bytes: byte_limits.pack, archive_limit: byte_limits.archive, - ci_logs: ci_logs.clone(), }) .with_maintenance(maintenance_handle.clone()) .with_limits(pack_limits) @@ -626,7 +617,6 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { secrets, entropy: Arc::new(OsEntropy), invite_codes: Default::default(), - ci_logs, admins, admission, contribution_policy, diff --git a/knot2/crates/knot-sim/src/harness.rs b/knot2/crates/knot-sim/src/harness.rs index 742b7715d..91b8781ab 100644 --- a/knot2/crates/knot-sim/src/harness.rs +++ b/knot2/crates/knot-sim/src/harness.rs @@ -800,7 +800,6 @@ fn assemble_router(parts: StateParts) -> (Router, FakeState) { secrets, entropy, invite_codes: Default::default(), - ci_logs: None, admins, admission, contribution_policy: knot_types::ContributionPolicy::Anyone, diff --git a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs index 493b23f93..97277763b 100644 --- a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs +++ b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs @@ -352,7 +352,6 @@ async fn spawn(published_line: String, with_h3: bool) -> World { admission: AdmissionPolicy::Closed, max_pack_bytes: knot_xrpc::MaxWireBytes::new(1 << 30), archive_limit: knot_git::ArchiveLimit::default(), - ci_logs: None, }) .with_lfs(lfs.clone(), 16), ); @@ -367,7 +366,6 @@ async fn spawn(published_line: String, with_h3: bool) -> World { let xrpc_state = Arc::new(knot_xrpc::XrpcState { imports, handover_sessions: Arc::new(knot_xrpc::handover::HandoverSessions::default()), - ci_logs: None, layout: layout.clone(), index: Arc::clone(&index), atproto, diff --git a/knot2/crates/knot-sim/tests/ssh.rs b/knot2/crates/knot-sim/tests/ssh.rs index f281ac0d8..94eff7e58 100644 --- a/knot2/crates/knot-sim/tests/ssh.rs +++ b/knot2/crates/knot-sim/tests/ssh.rs @@ -202,7 +202,6 @@ async fn spawn(published_line: String) -> Server { admission: knot_types::AdmissionPolicy::Closed, max_pack_bytes: knot_xrpc::MaxWireBytes::new(1 << 30), archive_limit: knot_git::ArchiveLimit::default(), - ci_logs: None, })); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let port = listener.local_addr().unwrap().port(); diff --git a/knot2/crates/knot-ssh/examples/ephemeral_knot.rs b/knot2/crates/knot-ssh/examples/ephemeral_knot.rs index 5c3104fe2..3dc634538 100644 --- a/knot2/crates/knot-ssh/examples/ephemeral_knot.rs +++ b/knot2/crates/knot-ssh/examples/ephemeral_knot.rs @@ -218,7 +218,6 @@ async fn main() { admission: AdmissionPolicy::Closed, max_pack_bytes: knot_pack::MaxWireBytes::new(1 << 34), archive_limit: knot_git::ArchiveLimit::default(), - ci_logs: None, })); let listener = TcpListener::bind(("127.0.0.1", port)).await.unwrap(); diff --git a/knot2/crates/knot-ssh/src/exec.rs b/knot2/crates/knot-ssh/src/exec.rs index f49ea360a..787752fb5 100644 --- a/knot2/crates/knot-ssh/src/exec.rs +++ b/knot2/crates/knot-ssh/src/exec.rs @@ -719,7 +719,6 @@ async fn serve_receive( maintenance: &state.maintenance, hostname: &state.hostname, catalog: Arc::clone(&state.catalog), - ci_logs: state.ci_logs.clone(), ref_surface: state.ref_surface.clone(), }) .await; diff --git a/knot2/crates/knot-ssh/src/lib.rs b/knot2/crates/knot-ssh/src/lib.rs index 52c307fa8..fa2efbe0d 100644 --- a/knot2/crates/knot-ssh/src/lib.rs +++ b/knot2/crates/knot-ssh/src/lib.rs @@ -17,7 +17,7 @@ use knot_maintenance::MaintenanceHandle; use knot_pack::{MaxWireBytes, PackLimits}; use knot_runtime::{Clock, Entropy, HttpTransport, OsEntropy}; use knot_types::{ - AccountDid, ActorId, AdmissionPolicy, AppviewEndpoint, CiLogsAddr, KnotHostname, TrustedProxies, + AccountDid, ActorId, AdmissionPolicy, AppviewEndpoint, KnotHostname, TrustedProxies, }; use russh::keys::ssh_key::rand_core; use russh::keys::{Algorithm, PrivateKey, ssh_key}; @@ -69,7 +69,6 @@ pub struct SshState { limits: PackLimits, max_pack_bytes: MaxWireBytes, archive_limit: ArchiveLimit, - ci_logs: Option, slots: Slots, lookup_slots: ResolveSlots, lookup_peers: Arc, @@ -102,7 +101,6 @@ pub struct SshConfig { pub admission: AdmissionPolicy, pub max_pack_bytes: MaxWireBytes, pub archive_limit: ArchiveLimit, - pub ci_logs: Option, } impl SshState { @@ -118,7 +116,6 @@ impl SshState { admission, max_pack_bytes, archive_limit, - ci_logs, } = config; Self { layout, @@ -132,7 +129,6 @@ impl SshState { limits: PackLimits::default(), max_pack_bytes, archive_limit, - ci_logs, slots: Slots::for_machine(), lookup_slots: ResolveSlots::new(MAX_INFLIGHT_LOOKUPS), lookup_peers: Arc::new(PreAuthLimiter::with_config(LimitConfig { diff --git a/knot2/crates/knot-ssh/tests/ssh_push.rs b/knot2/crates/knot-ssh/tests/ssh_push.rs index db4421751..7e5b23a91 100644 --- a/knot2/crates/knot-ssh/tests/ssh_push.rs +++ b/knot2/crates/knot-ssh/tests/ssh_push.rs @@ -429,7 +429,6 @@ async fn spawn_server_core( secrets, entropy: Arc::new(knot_runtime::OsEntropy), invite_codes: Default::default(), - ci_logs: None, admins: std::collections::BTreeSet::new(), admission: knot_types::AdmissionPolicy::Closed, contribution_policy: knot_types::ContributionPolicy::Anyone, @@ -500,7 +499,6 @@ async fn spawn_server_core( admission: knot_types::AdmissionPolicy::Closed, max_pack_bytes, archive_limit, - ci_logs: None, }); let base = match xrpc_state { Some(xrpc) => base.with_ref_surface(Arc::new(knot_xrpc::RefProjection { state: xrpc })), @@ -1367,7 +1365,6 @@ async fn launch(host_key_dir: &Path, layout: Layout, index: Arc, accounts admission: knot_types::AdmissionPolicy::Closed, max_pack_bytes: MaxWireBytes::new(1 << 30), archive_limit: ArchiveLimit::default(), - ci_logs: None, })); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let port = listener.local_addr().unwrap().port(); diff --git a/knot2/crates/knot-types/src/ids.rs b/knot2/crates/knot-types/src/ids.rs index 1a09cd1b8..03548232e 100644 --- a/knot2/crates/knot-types/src/ids.rs +++ b/knot2/crates/knot-types/src/ids.rs @@ -309,24 +309,6 @@ fn parse_knot_hostname(value: &str) -> Option { is_bare_host(value).then(|| KnotHostname(value.to_string())) } -fn parse_logs_host(value: &str) -> Option { - is_bare_host(value).then(|| LogsHost(value.to_string())) -} - -fn parse_ci_logs_addr(value: &str) -> Option { - let url = url::Url::parse(&format!("ssh://{value}")).ok()?; - (url.path().is_empty() - && url.username().is_empty() - && url.password().is_none() - && url.query().is_none() - && url.fragment().is_none()) - .then_some(())?; - Some(CiLogsAddr { - host: LogsHost::new(url.host_str()?).ok()?, - port: LogsPort::new(url.port()?)?, - }) -} - fn parse_branch_name(value: &str) -> Option { (value.len() <= 244 && RefName::new(format!("refs/heads/{value}")).is_ok()) .then(|| BranchName(value.to_string())) @@ -432,7 +414,6 @@ string_id!(TagName, "tag name", parse_tag_name); string_id!(RepoPath, "repository path", parse_repo_path); string_id!(AppviewEndpoint, "appview endpoint", parse_appview_endpoint); string_id!(KnotServiceUrl, "knot service URL", parse_service_url); -string_id!(LogsHost, "ci logs host", parse_logs_host); string_id!(PushOption, "push option", parse_push_option); impl RepoDid { @@ -513,51 +494,6 @@ impl PushOptions { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct LogsPort(std::num::NonZeroU16); - -impl LogsPort { - pub fn new(value: u16) -> Option { - std::num::NonZeroU16::new(value).map(Self) - } - - pub fn get(self) -> u16 { - self.0.get() - } -} - -impl fmt::Display for LogsPort { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - fmt::Display::fmt(&self.0, f) - } -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct CiLogsAddr { - host: LogsHost, - port: LogsPort, -} - -impl CiLogsAddr { - pub fn new(value: &str) -> Result { - parse_ci_logs_addr(value).ok_or_else(|| ParseError::invalid("ci logs address", value)) - } - - pub fn host(&self) -> &LogsHost { - &self.host - } - - pub fn port(&self) -> LogsPort { - self.port - } -} - -impl fmt::Display for CiLogsAddr { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}:{}", self.host, self.port) - } -} - #[derive(Clone, PartialEq, Eq)] pub enum OwnerRef { Did(OwnerDid), @@ -1438,33 +1374,6 @@ mod tests { ); } - #[test] - fn a_ci_logs_address_splits_into_a_bare_host_and_a_nonzero_port() { - let addr = CiLogsAddr::new("logs.oyster.cafe:3333").unwrap(); - assert_eq!(addr.host().as_str(), "logs.oyster.cafe"); - assert_eq!(addr.port().get(), 3333); - assert_eq!(addr.to_string(), "logs.oyster.cafe:3333"); - - [ - "logs.oyster.cafe", - "logs.oyster.cafe:0", - ":3333", - "::1", - "logs.oyster.cafe:+3333", - "logs.oyster.cafe:65536", - "logs.oyster.cafe: 3333", - "logs.oyster.cafe:33:33", - "logs.oyster.cafe:3333/logs", - "nel@logs.oyster.cafe:3333", - "logs.oyster.cafe:3333?tail=1", - "logs.oyster.cafe:3333#tail", - ] - .into_iter() - .for_each(|value| { - assert!(CiLogsAddr::new(value).is_err(), "{value}"); - }); - } - #[test] fn push_options_enforce_the_lexicon_bounds_at_construction() { assert_eq!( diff --git a/knot2/crates/knot-types/src/lib.rs b/knot2/crates/knot-types/src/lib.rs index f309eb046..4cda71ea1 100644 --- a/knot2/crates/knot-types/src/lib.rs +++ b/knot2/crates/knot-types/src/lib.rs @@ -6,9 +6,9 @@ pub use changes::{ChangedFiles, ChangedFilesBudget, Listing}; mod ids; pub use ids::{ - AccountDid, ActorId, AppviewEndpoint, AuthorName, BranchName, ChangeId, CiLogsAddr, ClonePath, + AccountDid, ActorId, AppviewEndpoint, AuthorName, BranchName, ChangeId, ClonePath, CobId, DidRkey, Email, HttpStatus, KnotHostname, KnotId, KnotServiceUrl, LabelKey, - LanguageBytes, LanguageName, LogsHost, LogsPort, ObjectCount, ObjectFormat, OfferedKey, Oid, + LanguageBytes, LanguageName, ObjectCount, ObjectFormat, OfferedKey, Oid, OriginUrl, OwnerDid, OwnerRef, ParseError, PushOption, PushOptions, RecordCollection, RecordRkey, RefName, RefTransition, RepoDid, RepoName, RepoPath, RepoRkey, ServiceDid, TagName, TypeName, UnixMicros, UnixSeconds, UriRkey, diff --git a/knot2/crates/knot-workflow/src/lib.rs b/knot2/crates/knot-workflow/src/lib.rs index eda71e752..5027163d0 100644 --- a/knot2/crates/knot-workflow/src/lib.rs +++ b/knot2/crates/knot-workflow/src/lib.rs @@ -98,16 +98,9 @@ impl Diagnostics { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] -pub enum PathMatch { - Assumed, - Listed, -} - #[derive(Debug, Clone, PartialEq, Eq)] pub struct CompiledWorkflow { pub name: WorkflowName, - pub paths: PathMatch, } pub struct Compiled { @@ -115,14 +108,6 @@ pub struct Compiled { pub diagnostics: Diagnostics, } -impl Compiled { - pub fn any_listed_match(&self) -> bool { - self.workflows - .iter() - .any(|workflow| workflow.paths == PathMatch::Listed) - } -} - pub fn compile(raw: &[RawWorkflow], trigger: &Trigger, changed: &ChangedFiles) -> Compiled { let (workflows, diagnostics) = raw .iter() @@ -163,14 +148,14 @@ fn compile_one( return (None, diag); } }; - let Some(paths) = matched else { + if !matched { diag.warning( raw.name.as_str(), "workflow skipped", &format!("didn't match trigger {}", trigger.kind()), ); return (None, diag); - }; + } let depth = match CloneDepth::parse(parsed.clone.depth) { Ok(depth) => depth, Err(error) => { @@ -187,7 +172,6 @@ fn compile_one( Ok(()) => ( Some(CompiledWorkflow { name: raw.name.clone(), - paths, }), diag, ), @@ -222,21 +206,21 @@ fn workflow_matches( when: &[Constraint], trigger: &Trigger, changed: &ChangedFiles, -) -> Result, String> { +) -> Result { if when.is_empty() { - return Ok(Some(PathMatch::Listed)); + return Ok(true); } when.iter() .map(|constraint| constraint_matches(constraint, trigger, changed)) - .collect::>, String>>() - .map(|results| results.into_iter().flatten().max()) + .collect::, String>>() + .map(|results| results.contains(&true)) } fn constraint_matches( constraint: &Constraint, trigger: &Trigger, changed: &ChangedFiles, -) -> Result, String> { +) -> Result { match trigger { Trigger::Push { ref_name } => { let event = constraint.event.0.iter().any(|kind| kind == "push"); @@ -253,12 +237,10 @@ fn constraint_matches( // We aren't the one deciding whether this workflow runs, remember, // spindles are, and spindles will decide from the // atproto record we will emit. - let paths = match (constraint.paths.0.is_empty(), listed, changed.listing()) { - (true, _, _) | (false, true, _) => Some(PathMatch::Listed), - (false, false, Listing::Truncated) => Some(PathMatch::Assumed), - (false, false, Listing::Complete) => None, - }; - Ok(paths.filter(|_| event && reference)) + let paths = constraint.paths.0.is_empty() + || listed + || matches!(changed.listing(), Listing::Truncated); + Ok(paths && event && reference) } } } @@ -402,10 +384,6 @@ mod tests { .collect::>(), vec!["test.yml"] ); - assert!( - compiled.any_listed_match(), - "a workflow with no paths constraint never depends on the listing" - ); assert!( compiled.diagnostics.is_empty(), "{:?}", @@ -449,7 +427,7 @@ mod tests { } #[test] - fn a_truncated_listing_assumes_a_paths_match_and_only_a_listed_hit_promises_the_run() { + fn a_truncated_listing_assumes_a_paths_match() { let yaml = "engine: nixery.dev/x\nwhen:\n - event: push\n branch: ['**']\n paths: ['src/**']\n"; assert_eq!( compile( @@ -473,39 +451,6 @@ mod tests { 1, "a listing the record couldn't hold rules no glob out" ); - assert_eq!(assumed.workflows[0].paths, PathMatch::Assumed); - assert!( - !assumed.any_listed_match(), - "spindle reads the same truncated listing and skips this run" - ); - - let mut budget = knot_types::ChangedFilesBudget::new(); - let _ = budget.admit(knot_types::RepoPath::new("src/deep/main.rs").unwrap()); - let _ = budget.truncate(); - let listed = compile( - &[raw("ci.yml", yaml)], - &push("refs/heads/main"), - &budget.finish(), - ); - assert_eq!(listed.workflows[0].paths, PathMatch::Listed); - assert!( - listed.any_listed_match(), - "spindle sees the same listed path and runs this workflow" - ); - - let strongest = compile( - &[raw( - "ci.yml", - "engine: nixery.dev/x\nwhen:\n - event: push\n branch: ['**']\n paths: ['never/**']\n - event: push\n branch: ['**']\n", - )], - &push("refs/heads/main"), - &ChangedFiles::unknown(), - ); - assert_eq!( - strongest.workflows[0].paths, - PathMatch::Listed, - "an assumed constraint never weakens an unconstrained one" - ); } #[test] diff --git a/knot2/crates/knot-xrpc/src/lib.rs b/knot2/crates/knot-xrpc/src/lib.rs index 158dca6b3..86ab69fe0 100644 --- a/knot2/crates/knot-xrpc/src/lib.rs +++ b/knot2/crates/knot-xrpc/src/lib.rs @@ -82,7 +82,7 @@ use knot_resource::Slots; use knot_runtime::{Clock, Entropy, HttpTransport}; use knot_secrets::SealedStore; use knot_types::{ - AccountDid, AdmissionPolicy, AppviewEndpoint, CiLogsAddr, ClonePath, ContributionPolicy, + AccountDid, AdmissionPolicy, AppviewEndpoint, ClonePath, ContributionPolicy, KnotHostname, KnotId, KnotServiceUrl, Nsid, OwnerDid, OwnerRef, RepoDid, UnixSeconds, }; @@ -194,7 +194,6 @@ pub struct XrpcState { pub knot_did: KnotId, pub knot_repo: RepoDid, pub knot_hostname: KnotHostname, - pub ci_logs: Option, pub meta_path: PathBuf, pub knot_service_url: KnotServiceUrl, pub writes: Arc, diff --git a/knot2/crates/knot-xrpc/src/outbox.rs b/knot2/crates/knot-xrpc/src/outbox.rs index 4d4524179..e26e3791d 100644 --- a/knot2/crates/knot-xrpc/src/outbox.rs +++ b/knot2/crates/knot-xrpc/src/outbox.rs @@ -8,7 +8,7 @@ use knot_resource::{FileMode, atomic_write_bytes}; use knot_runtime::{Clock, HttpTransport}; use knot_types::{RepoDid, ServiceDid, UnixSeconds}; use serde::{Deserialize, Serialize}; -use tokio::sync::Notify; +use tokio::sync::{Notify, oneshot}; use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; @@ -58,6 +58,8 @@ struct Inner { next_id: u64, backoff: HashMap, in_flight: HashSet, + // only an entry's first attempt answers, a retry lands after the push output is gone + pushers: HashMap>>, #[cfg(test)] writes: usize, } @@ -131,6 +133,7 @@ impl Outbox { next_id, backoff: HashMap::new(), in_flight: HashSet::new(), + pushers: HashMap::new(), #[cfg(test)] writes: 0, }), @@ -145,15 +148,20 @@ impl Outbox { repo: &RepoDid, events: Vec<(String, serde_json::Value)>, now: UnixSeconds, - ) { + ) -> Vec>> { if events.is_empty() { - return; + return Vec::new(); } let mut inner = self.inner.lock().expect("outbox lock isn't poisoned"); + inner.pushers.retain(|_, pusher| !pusher.is_closed()); + let mut replies = Vec::new(); for (ref_name, body) in events { self.make_room(&mut inner, spindle); let id = inner.next_id; inner.next_id += 1; + let (pusher, reply) = oneshot::channel(); + inner.pushers.insert(id, pusher); + replies.push(reply); inner.entries.push_back(Entry { id, spindle: spindle.clone(), @@ -167,6 +175,7 @@ impl Outbox { self.persist(&mut inner); drop(inner); self.wake.notify_one(); + replies } fn make_room(&self, inner: &mut Inner, spindle: &ServiceDid) { @@ -278,11 +287,14 @@ impl Outbox { } /// a claimed entry that never settles keeps its spindle blocked for good - fn settle(&self, entry: &Entry, result: Result<(), Failure>, now: Instant) { + fn settle(&self, entry: &Entry, result: Result, Failure>, now: Instant) { let mut inner = self.inner.lock().expect("outbox lock isn't poisoned"); inner.in_flight.remove(&entry.spindle); + if let Some(pusher) = inner.pushers.remove(&entry.id) { + let _ = pusher.send(result.as_ref().ok().cloned().flatten()); + } match result { - Ok(()) => { + Ok(_) => { inner.entries.retain(|queued| queued.id != entry.id); inner.backoff.remove(&entry.spindle); self.persist(&mut inner); @@ -342,20 +354,20 @@ pub(crate) async fn deliver( atproto: &Atproto, auth: &dyn PointerAuthorizer, entry: &Entry, -) -> Result<(), AtprotoError> { +) -> Result, AtprotoError> { let answer = atproto .notify_spindle(auth, &entry.spindle, entry.body.to_string().into()) .await?; - let pipeline = match &answer { - SpindleAnswer::Scheduled(id) => Some(id.as_str()), - SpindleAnswer::Nothing => None, + let (pipeline, message) = match answer { + SpindleAnswer::Scheduled { pipeline, message } => (Some(pipeline), message), + SpindleAnswer::Nothing => (None, None), SpindleAnswer::Malformed(problem) => { tracing::warn!( spindle = entry.spindle.as_str(), problem = problem.as_str(), "the spindle's answer to the push wasn't readable, counting it delivered" ); - None + (None, None) } }; tracing::info!( @@ -365,7 +377,7 @@ pub(crate) async fn deliver( pipeline, "the spindle heard of the push" ); - Ok(()) + Ok(message) } async fn attempt(state: Arc>, entry: Entry) { @@ -522,8 +534,9 @@ mod tests { let body: serde_json::Value = serde_json::from_slice(request.body.as_ref().unwrap()).unwrap(); let ref_name = body["event"]["ref"].as_str().unwrap().to_string(); + let answer = serde_json::json!({ "pipeline": "3kabc", "message": format!("follow {ref_name}") }); taken.lock().unwrap().push((host, ref_name)); - Ok(reply(200, "{}".to_string())) + Ok(reply(200, answer.to_string())) }) }; let atproto = Atproto::new( @@ -569,6 +582,51 @@ mod tests { .collect() } + #[tokio::test] + async fn the_pusher_hears_what_the_spindle_said_on_the_first_attempt() { + let dir = tempfile::tempdir().unwrap(); + let outbox = outbox(dir.path()); + let world = spindles(0, None); + let mut replies = outbox.enqueue_all( + &spindle(SPINDLE), + &RepoDid::new("did:plc:squid").unwrap(), + vec![( + "refs/heads/main".to_string(), + serde_json::json!({ "event": { "ref": "refs/heads/main" } }), + )], + UnixSeconds::new(START_SECS), + ); + pump(&outbox, &world, Instant::now()).await; + assert_eq!( + replies.remove(0).await.unwrap(), + Some("follow refs/heads/main".to_string()) + ); + assert_eq!(world.taken.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn a_failed_first_attempt_tells_the_pusher_nothing_and_the_retry_still_delivers_once() { + let dir = tempfile::tempdir().unwrap(); + let outbox = outbox(dir.path()); + let world = spindles(1, None); + let mut replies = outbox.enqueue_all( + &spindle(SPINDLE), + &RepoDid::new("did:plc:squid").unwrap(), + vec![( + "refs/heads/main".to_string(), + serde_json::json!({ "event": { "ref": "refs/heads/main" } }), + )], + UnixSeconds::new(START_SECS), + ); + let t0 = Instant::now(); + pump(&outbox, &world, t0).await; + assert_eq!(replies.remove(0).await.unwrap(), None); + pump(&outbox, &world, t0 + Duration::from_secs(1)).await; + assert_eq!(world.attempts.load(Ordering::SeqCst), 2); + assert_eq!(world.taken.lock().unwrap().len(), 1); + assert!(outbox.queued().is_empty()); + } + #[test] fn backoff_doubles_up_to_the_cap() { let delays: Vec = (1..=6) @@ -710,7 +768,7 @@ mod tests { refs(first.clone()), ["refs/heads/slow-1", "refs/heads/fast-1"] ); - outbox.settle(&first[1], Ok(()), t0); + outbox.settle(&first[1], Ok(None), t0); let second = outbox.claim_due(t0, wall); assert_eq!( diff --git a/knot2/crates/knot-xrpc/src/receive.rs b/knot2/crates/knot-xrpc/src/receive.rs index 0419349f3..b00bd350d 100644 --- a/knot2/crates/knot-xrpc/src/receive.rs +++ b/knot2/crates/knot-xrpc/src/receive.rs @@ -183,7 +183,6 @@ async fn serve_receive( maintenance: &state.maintenance, hostname: &state.knot_hostname, catalog: Arc::clone(&state.catalog), - ci_logs: state.ci_logs.clone(), ref_surface: Some(Arc::new(crate::RefProjection { state: Arc::clone(state), })), diff --git a/knot2/crates/knot-xrpc/src/refrecords.rs b/knot2/crates/knot-xrpc/src/refrecords.rs index bdf20eb1f..e50534b5f 100644 --- a/knot2/crates/knot-xrpc/src/refrecords.rs +++ b/knot2/crates/knot-xrpc/src/refrecords.rs @@ -124,10 +124,10 @@ impl knot_receive::RefSurface for RefProjection Vec { sync_refs(&self.state, repo, repo_did, applied); // queued, so it doesn't need the signer, the outbox signs per attempt - notify_spindle(&self.state, repo, repo_did, applied, pusher, options); + let replies = notify_spindle(&self.state, repo, repo_did, applied, pusher, options); match self.state.secrets.signer(&self.state.knot_did) { Ok(signer) => firehose::publish_push( &self.state.firehose, @@ -144,6 +144,7 @@ impl knot_receive::RefSurface for RefProjection( applied: &[RefUpdate], pusher: &AccountDid, options: &PushOptions, -) { +) -> Vec { let spindle = match get_repo_spindle(repo) { Ok(Some(spindle)) => spindle, - Ok(None) => return, + Ok(None) => return Vec::new(), Err(error) => { tracing::warn!(repo = did.as_str(), %error, "the manifest wouldn't read, so the spindle doesn't hear of this push"); - return; + return Vec::new(); } }; let events: Vec<_> = applied @@ -171,7 +172,7 @@ fn notify_spindle( Some((update.name().as_str().to_string(), body)) }) .collect(); - state.outbox.enqueue_all(&spindle, did, events, state.now()); + state.outbox.enqueue_all(&spindle, did, events, state.now()) } /// real events are a few hundred bytes, so this only bites ref names and push options built to bloat the queue diff --git a/knot2/crates/knot-xrpc/src/tests.rs b/knot2/crates/knot-xrpc/src/tests.rs index e18fa927a..a40f1b059 100644 --- a/knot2/crates/knot-xrpc/src/tests.rs +++ b/knot2/crates/knot-xrpc/src/tests.rs @@ -355,7 +355,6 @@ fn state_from( secrets, entropy: Arc::new(OsEntropy), invite_codes: Default::default(), - ci_logs: None, admins: BTreeSet::from([account(ADMIN_HOST)]), admission, contribution_policy: knot_types::ContributionPolicy::Anyone, diff --git a/knot2/crates/knot-xrpc/tests/common/mod.rs b/knot2/crates/knot-xrpc/tests/common/mod.rs index 36a0f70d1..10d5818a5 100644 --- a/knot2/crates/knot-xrpc/tests/common/mod.rs +++ b/knot2/crates/knot-xrpc/tests/common/mod.rs @@ -691,7 +691,6 @@ impl World { secrets, entropy: Arc::new(OsEntropy), invite_codes: Default::default(), - ci_logs: None, admins, admission: knot_types::AdmissionPolicy::Closed, contribution_policy: policy, diff --git a/knot2/example.toml b/knot2/example.toml index a63b91916..7083e2af6 100644 --- a/knot2/example.toml +++ b/knot2/example.toml @@ -490,10 +490,6 @@ # Can also be specified via environment variable `KNOT_HOMEPAGE_PATH`. #path = -[ci] -# Can also be specified via environment variable `KNOT_CI_LOGS_ADDR`. -#logs_addr = - [messages] [messages.push] # Can also be specified via environment variable `KNOT_MESSAGES_PUSH_ACK`. @@ -512,10 +508,6 @@ # Default value: ["no pipelines to compile"] #pipeline_none = ["no pipelines to compile"] -# Can also be specified via environment variable `KNOT_MESSAGES_PUSH_CI_LOGS`. -# Default value: ["-> Browse CI logs in your terminal:", " ssh -t -p {port} {host} {repo} {sha}"] -#ci_logs = ["-> Browse CI logs in your terminal:", " ssh -t -p {port} {host} {repo} {sha}"] - [messages.fetch] # Can also be specified via environment variable `KNOT_MESSAGES_FETCH_MOTD`. # Default value: ["Thanks for using {knot}!"] -- knotmirror 0.0.1