the drain timer can activate, and stops rebuilding a handles publish that failed
Signed-off-by: dawn <90008@klbr.net>
51 unmodified lines5253545556575859606162636465666768697071727374757677787980818283848586878889909192939495965 unmodified lines51 unmodified lines doCheck = false; meta.mainProgram = "knot-static-firehose"; }; packages.default = self.packages.${system}.knot-static; # every unit that can activate runs the reload command and git from its own PATH checks = pkgs.lib.optionalAttrs pkgs.stdenv.hostPlatform.isLinux { module-units-can-activate = let nixos = nixpkgs.lib.nixosSystem { inherit system; modules = [ self.nixosModules.knot-static { services.knot-static = { enable = true; hostname = "knot.test"; owner = "did:plc:owner"; }; boot.loader.grub.enable = false; fileSystems."/" = { device = "none"; fsType = "tmpfs"; }; system.stateVersion = "25.11"; } ]; }; conf = nixos.config.environment.etc."knot-static/config.toml".source; units = [ "knot-static-cgi" "knot-static-drain" "knot-static-prepare" ]; check = unit: '' echo "checking ${unit}" env -i PATH='${nixos.config.systemd.services.${unit}.environment.PATH}' sh -c 'command -v git' >/dev/null env -i PATH='${nixos.config.systemd.services.${unit}.environment.PATH}' sh -c "$reload" ''; in pkgs.runCommand "knot-static-module-units-can-activate" { nativeBuildInputs = [ pkgs.python3 ]; } '' reload=$(python3 -c 'import sys, tomllib; print(tomllib.load(open(sys.argv[1], "rb"))["activate"]["reload_command"])' ${conf}) ${pkgs.lib.concatMapStrings check units} touch $out ''; }; devShells.default = pkgs.mkShell { packages = [ (fenix.packages.${system}.stable.toolchain) pkgs.cmake pkgs.perl pkgs.pkg-config pkgs.gitoxide pkgs.cargo-nextest ]; }; })5 unmodified lines25 unmodified lines26272829303132333435363738248 unmodified linesExpand all28728828929028628728828929029129229329429542 unmodified lines33833934034134134234334434534634717 unmodified lines365366367368367368369370371369370371372373101 unmodified linesExpand all25 unmodified lines firehoseSocket = "/run/knot-static-firehose.sock"; # generations name these paths, so they must outlive a rebuild configPath = "/etc/knot-static/config.toml"; binPath = "/run/current-system/sw/bin/knot-static"; # what an activation spawns: reload_command through sh, git for the exports activatePath = [ pkgs.gitMinimal pkgs.bash ];
defaultSettings = { knot = { inherit (cfg) hostname owner;248 unmodified linesExpand all systemd.services.knot-static-cgi = { description = "fcgiwrap for knot-static's cob writes"; after = [ "network.target" ]; wantedBy = [ "multi-user.target" ]; # activate runs reload_command through sh path = [ pkgs.gitMinimal pkgs.bash ]; path = activatePath; environment.KNOT_STATIC_CONFIG = configPath; unitConfig.ConditionPathExists = cfg.environmentFile; serviceConfig = { User = cfg.user;42 unmodified lines
# pushes deliver spindle events themselves, this retries the ones a spindle missed systemd.services.knot-static-drain = { description = "deliver knot-static's queued spindle events"; path = [ pkgs.gitMinimal ]; # post-activate publishes refreshed handles path = activatePath; unitConfig.ConditionPathExists = cfg.environmentFile; serviceConfig = { Type = "oneshot"; User = cfg.user;17 unmodified lines restartTriggers = [ cfg.package config.environment.etc."knot-static/config.toml".source ]; # activate runs reload_command through sh path = [ pkgs.gitMinimal pkgs.bash ]; path = activatePath; unitConfig.ConditionPathExists = cfg.environmentFile; serviceConfig = { Type = "oneshot"; RemainAfterExit = true;101 unmodified linesExpand all1128 unmodified linesExpand all112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811128 unmodified linesExpand all value["built_by"] = built_by; rewrite(&value); assert!(unchanged(), "and only once");}
#[test]fn a_handles_publish_that_failed_waits_for_a_change_before_it_rebuilds_again() { let env = setup_test_env(); assert!(knot_static(&env.config, ["generate"]).status.success()); let first = get_only_generation_dir(&env.output); let first_id = first.file_name().unwrap().to_str().unwrap().to_string(); assert!( knot_static(&env.config, ["activate", &first_id]) .status .success() ); let config = std::fs::read_to_string(&env.config).unwrap(); std::fs::write( &env.config, config.replace( "nginx_binary = \"true\"\n", "nginx_binary = \"true\"\nreload_command = \"exit 1\"\n", ), ) .unwrap(); let profile = |description: &str| { std::fs::write( env.output.join("owner-handles.json"), serde_json::to_vec(&serde_json::json!({ "version": 1, "entries": { OWNER: { "handle": "ptr.pet", "verified_at_unix": 1_700_000_000u64, "profile": { "description": description }, } } })) .unwrap(), ) .unwrap(); }; let tried = || { let drained = knot_static(&env.config, ["post-activate"]); String::from_utf8_lossy(&drained.stderr).contains("publish the refreshed handles") }; profile("one"); assert!(tried(), "the first drain tries and its activation fails"); assert!(!tried(), "the next drain doesn't rebuild the same thing"); profile("two"); assert!(tried(), "a different store tries again"); assert_eq!(active_generation(&env.output), first_id);}162 unmodified linesExpand all163164165166167168169170167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219363 unmodified linesExpand all162 unmodified linesExpand all refresh_handles_with(&atproto, &pds, &avatars, &named, &mut stored, now).await; if stored != before && record.put(&stored).is_ok() { sweep_avatars(&avatars, &stored); } if manifest.handles_digest.as_deref() != Some(served_digest(&stored).as_str()) && let Err(error) = crate::writes::publish_scoped(config, None).await { tracing::warn!("publish the refreshed handles: {error:#}"); let digest = served_digest(&stored); if manifest.handles_digest.as_deref() == Some(digest.as_str()) { return; } let attempt = FailedPublish { digest, generation: manifest.generation_id.clone(), at_unix: now, }; let failed_path = config.output_root.join(FAILED_PUBLISH_FILE); let failed = std::fs::read(&failed_path) .ok() .and_then(|bytes| serde_json::from_slice::<FailedPublish>(&bytes).ok()); if failed.is_some_and(|failed| failed.holds_back(&attempt)) { return; } match crate::writes::publish_scoped(config, None).await { Ok(_) => { let _ = std::fs::remove_file(&failed_path); } Err(error) => { tracing::warn!("publish the refreshed handles: {error:#}"); if let Ok(bytes) = serde_json::to_vec(&attempt) { let _ = crate::cas::atomic_write(&failed_path, &bytes); } } }}
const FAILED_PUBLISH_FILE: &str = "handles-publish.failed.json";/// a publish that failed for these handles over this generation isn't tried again soonerconst PUBLISH_RETRY_SECS: u64 = 3600;
/// the last handles publish that failed, so the drain timer doesn't rebuild every run#[derive(Debug, PartialEq, Serialize, Deserialize)]pub(crate) struct FailedPublish { pub(crate) digest: String, pub(crate) generation: String, pub(crate) at_unix: u64,}
impl FailedPublish { pub(crate) fn holds_back(&self, attempt: &FailedPublish) -> bool { self.digest == attempt.digest && self.generation == attempt.generation && attempt.at_unix.saturating_sub(self.at_unix) < PUBLISH_RETRY_SECS }}
fn refresh_atproto( config: &ValidatedConfig,) -> Option<Atproto<ReqwestHttp, knot_runtime::SystemClock>> { config363 unmodified linesExpand all1232345671093 unmodified linesExpand all1101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131use super::handles::{ AUTHOR_REVERIFY_SECS, Fetched, Named, StoredHandle, profile_of, published_description, refresh_handles_with, verify_unseen_with, AUTHOR_REVERIFY_SECS, FailedPublish, Fetched, Named, StoredHandle, profile_of, published_description, refresh_handles_with, verify_unseen_with,};use super::*;use crate::config::IndexConfig;use crate::datastore::{Datastore, schema};1093 unmodified linesExpand all assert!(row_has(240, [0xf1, 0xe4, 0xea]), "title ink"); assert!(row_has(310, [0xf2, 0x8b, 0xb5]), "accent ink"); assert!(row_has(390, [0xa8, 0x91, 0x9d]), "description ink");}
#[test]fn a_failed_handles_publish_holds_back_only_the_same_attempt_for_a_while() { let failed = FailedPublish { digest: "d".into(), generation: "g".into(), at_unix: 1000, }; let attempt = |digest: &str, generation: &str, at_unix| FailedPublish { digest: digest.into(), generation: generation.into(), at_unix, }; assert!(failed.holds_back(&attempt("d", "g", 1300))); assert!( !failed.holds_back(&attempt("e", "g", 1300)), "the store changed" ); assert!( !failed.holds_back(&attempt("d", "h", 1300)), "another generation is active" ); assert!( !failed.holds_back(&attempt("d", "g", 1000 + 3600)), "an hour later" );}