GameCult/Odin
Watch
9
0
Fork
You've already forked Odin
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Make Idunn admission atomic and request-bound

Browse source
This commit is contained in:
Meta Vens 2026-07-17 02:37:42 +01:00
commit 92fab01d37
10 changed files with 1472 additions and 112 deletions
Download patch file Download diff file Expand all files Collapse all files

127
Cargo.lock generated
View file

@ -64,6 +64,12 @@ version = "0.22.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
[[package]]
name = "base64ct"
version = "1.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06"
[[package]]
name = "bitflags"
version = "2.12.1"
@ -125,6 +131,12 @@ dependencies = [
"inout",
]
[[package]]
name = "const-oid"
version = "0.9.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
[[package]]
name = "core-foundation-sys"
version = "0.8.7"
@ -173,6 +185,7 @@ dependencies = [
"serde_bytes",
"serde_json",
"uuid",
"windows-sys 0.61.2",
]
[[package]]
@ -215,6 +228,43 @@ dependencies = [
"uuid",
]
[[package]]
name = "curve25519-dalek"
version = "4.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be"
dependencies = [
"cfg-if",
"cpufeatures",
"curve25519-dalek-derive",
"digest",
"fiat-crypto",
"rustc_version",
"subtle",
"zeroize",
]
[[package]]
name = "curve25519-dalek-derive"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "der"
version = "0.7.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb"
dependencies = [
"const-oid",
"zeroize",
]
[[package]]
name = "diff"
version = "0.1.13"
@ -232,6 +282,30 @@ dependencies = [
"subtle",
]
[[package]]
name = "ed25519"
version = "2.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53"
dependencies = [
"pkcs8",
"signature",
]
[[package]]
name = "ed25519-dalek"
version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9"
dependencies = [
"curve25519-dalek",
"ed25519",
"serde",
"sha2",
"subtle",
"zeroize",
]
[[package]]
name = "equivalent"
version = "1.0.2"
@ -254,6 +328,12 @@ version = "2.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6"
[[package]]
name = "fiat-crypto"
version = "0.2.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d"
[[package]]
name = "find-msvc-tools"
version = "0.1.9"
@ -425,10 +505,13 @@ dependencies = [
"cultcache-rs",
"cultmesh-rs",
"cultnet-rs",
"ed25519-dalek",
"odin-core",
"rmp-serde",
"serde",
"sha2",
"tempfile",
"uuid",
]
[[package]]
@ -568,6 +651,16 @@ version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
[[package]]
name = "pkcs8"
version = "0.10.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7"
dependencies = [
"der",
"spki",
]
[[package]]
name = "polyval"
version = "0.6.2"
@ -707,6 +800,15 @@ dependencies = [
"serde_bytes",
]
[[package]]
name = "rustc_version"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92"
dependencies = [
"semver",
]
[[package]]
name = "rustix"
version = "1.1.4"
@ -802,6 +904,15 @@ version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"
[[package]]
name = "signature"
version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de"
dependencies = [
"rand_core 0.6.4",
]
[[package]]
name = "slab"
version = "0.4.12"
@ -832,6 +943,16 @@ dependencies = [
"windows-sys 0.52.0",
]
[[package]]
name = "spki"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d"
dependencies = [
"base64ct",
"der",
]
[[package]]
name = "subtle"
version = "2.6.1"
@ -1311,6 +1432,12 @@ dependencies = [
"syn",
]
[[package]]
name = "zeroize"
version = "1.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e"
[[package]]
name = "zmij"
version = "1.0.21"

5
crates/idunn-daemon/Cargo.toml
View file

@ -19,3 +19,8 @@ rmp-serde = "1.3"
sha2 = "0.10"
chrono = "0.4"
serde = { version = "1", features = ["derive"] }
ed25519-dalek = "2.2"
uuid = "1"
[dev-dependencies]
tempfile = "3"

1136
crates/idunn-daemon/src/main.rs
View file

File diff suppressed because it is too large Load diff

168
crates/odin-core/src/documents.rs
View file

@ -13,10 +13,12 @@ pub const EVE_PROVIDER_ADVERTISEMENT_SCHEMA: &str = "gamecult.eve.provider_adver
pub const VOIDBOT_SWARM_STATE_SNAPSHOT_SCHEMA: &str = "voidbot.swarm_state_snapshot.v1";
pub const IDUNN_DESIRED_DAEMON_SCHEMA: &str = "idunn.desired_daemon.v1";
pub const IDUNN_DAEMON_HEALTH_SCHEMA: &str = "idunn.daemon_health.v1";
pub const IDUNN_SIGNED_HEALTH_ADMISSION_SCHEMA: &str = "idunn.signed_health_admission.v1";
pub const IDUNN_KEEPALIVE_DECISION_SCHEMA: &str = "idunn.keepalive_decision.v1";
pub const IDUNN_RESTART_REQUEST_SCHEMA: &str = "idunn.restart_request.v1";
pub const IDUNN_RESTART_RESULT_SCHEMA: &str = "idunn.restart_result.v1";
pub const IDUNN_DEPLOYMENT_REQUEST_SCHEMA: &str = "idunn.deployment_request.v2";
pub const IDUNN_CURRENT_DEPLOYMENT_REQUEST_SCHEMA: &str = "idunn.current_deployment_request.v1";
pub const IDUNN_DEPLOYMENT_RESULT_SCHEMA: &str = "idunn.deployment_result.v1";
pub const IDUNN_LIFECYCLE_COMMAND_SCHEMA: &str = "idunn.lifecycle_command.v1";
pub const IDUNN_RELEASE_TARGET_SCHEMA: &str = "idunn.release_target.v2";
@ -49,7 +51,8 @@ pub const MUNINN_MOVE_IDENTITY_SCHEMA: &str = "muninn.move_identity.v1";
pub const MUNINN_MOVE_LIGHT_COMMAND_SCHEMA: &str = "muninn.move_light_command.v1";
pub const MUNINN_MOVE_HUE_PROGRAM_SCHEMA: &str = "muninn.move_hue_program.v1";
pub const MUNINN_MOVE_TRACKER_HEALTH_SCHEMA: &str = "muninn.move_tracker_health.v1";
pub const MUNINN_MOVE_EVIDENCE_TRANSPORT_HEALTH_SCHEMA: &str = "muninn.move_evidence_transport_health.v1";
pub const MUNINN_MOVE_EVIDENCE_TRANSPORT_HEALTH_SCHEMA: &str =
"muninn.move_evidence_transport_health.v1";
pub const MUNINN_QUEST_ACCESS_SCHEMA: &str = "muninn.quest_access.v1";
pub const MUNINN_COMMAND_BOUNDARY_SCHEMA: &str = "muninn.command_boundary.v1";
pub const MUNINN_TRANSPORT_PROFILE_SCHEMA: &str = "muninn.transport_profile.v1";
@ -270,6 +273,42 @@ pub struct IdunnDaemonHealthRecord {
pub transport: String,
}
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(
type = "idunn.signed_health_admission",
schema = "idunn.signed_health_admission.v1"
)]
pub struct IdunnSignedHealthAdmissionRecord {
#[cultcache(key = 0)]
pub daemon_id: String,
#[cultcache(key = 1)]
pub state: String,
#[cultcache(key = 2)]
pub observed_at: String,
#[cultcache(key = 3)]
pub admitted_at: String,
#[cultcache(key = 4)]
pub health_contract: String,
#[cultcache(key = 5)]
pub deployment_request_id: String,
#[cultcache(key = 6)]
pub release_id: String,
#[cultcache(key = 7)]
pub release_witness_sha256: String,
#[cultcache(key = 8)]
pub source_commit: String,
#[cultcache(key = 9)]
pub publisher_incarnation_id: String,
#[cultcache(key = 10)]
pub publisher_sequence: u64,
#[cultcache(key = 11)]
pub publisher_process_created_at: String,
#[cultcache(key = 12)]
pub signer_identity_id: String,
#[cultcache(key = 13)]
pub signed_health_sha256: String,
}
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(
type = "idunn.keepalive_decision",
@ -352,6 +391,22 @@ pub struct IdunnDeploymentRequestRecord {
pub requires_bifrost_authority: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(
type = "idunn.current_deployment_request",
schema = "idunn.current_deployment_request.v1"
)]
pub struct IdunnCurrentDeploymentRequestRecord {
#[cultcache(key = 0)]
pub daemon_id: String,
#[cultcache(key = 1)]
pub request_id: String,
#[cultcache(key = 2)]
pub sequence: u64,
#[cultcache(key = 3)]
pub updated_at: String,
}
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(
type = "idunn.deployment_result",
@ -1286,46 +1341,85 @@ pub struct MuninnMoveHueProgramRecord {
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "muninn.move_tracker_health", schema = "muninn.move_tracker_health.v1")]
#[cultcache(
type = "muninn.move_tracker_health",
schema = "muninn.move_tracker_health.v1"
)]
pub struct MuninnMoveTrackerHealthRecord {
#[cultcache(key = 0)] pub health_id: String,
#[cultcache(key = 1)] pub host_id: String,
#[cultcache(key = 2)] pub camera_id: String,
#[cultcache(key = 3)] pub camera_index: i32,
#[cultcache(key = 4)] pub state: String,
#[cultcache(key = 5)] pub camera_name: String,
#[cultcache(key = 6)] pub camera_api: String,
#[cultcache(key = 7)] pub width: u32,
#[cultcache(key = 8)] pub height: u32,
#[cultcache(key = 9)] pub exposure: f32,
#[cultcache(key = 10)] pub calibrated_controller_count: u32,
#[cultcache(key = 11)] pub update_count: u64,
#[cultcache(key = 12)] pub observation_count: u64,
#[cultcache(key = 13)] pub latest_observation_count: u32,
#[cultcache(key = 14)] pub last_observation_at: String,
#[cultcache(key = 15)] pub detail: String,
#[cultcache(key = 16)] pub updated_at: String,
#[cultcache(key = 17, default)] pub image_mean_rgb: Vec<u32>,
#[cultcache(key = 18, default)] pub image_peak_rgb: Vec<u32>,
#[cultcache(key = 19, default)] pub color_evidence_move_ids: Vec<String>,
#[cultcache(key = 20, default)] pub color_evidence_pixel_counts: Vec<u32>,
#[cultcache(key = 21, default)] pub rejected_stale_count: u64,
#[cultcache(key = 22, default)] pub rejected_radius_count: u64,
#[cultcache(key = 23, default)] pub rejected_bounds_count: u64,
#[cultcache(key = 24, default)] pub rejected_continuity_count: u64,
#[cultcache(key = 0)]
pub health_id: String,
#[cultcache(key = 1)]
pub host_id: String,
#[cultcache(key = 2)]
pub camera_id: String,
#[cultcache(key = 3)]
pub camera_index: i32,
#[cultcache(key = 4)]
pub state: String,
#[cultcache(key = 5)]
pub camera_name: String,
#[cultcache(key = 6)]
pub camera_api: String,
#[cultcache(key = 7)]
pub width: u32,
#[cultcache(key = 8)]
pub height: u32,
#[cultcache(key = 9)]
pub exposure: f32,
#[cultcache(key = 10)]
pub calibrated_controller_count: u32,
#[cultcache(key = 11)]
pub update_count: u64,
#[cultcache(key = 12)]
pub observation_count: u64,
#[cultcache(key = 13)]
pub latest_observation_count: u32,
#[cultcache(key = 14)]
pub last_observation_at: String,
#[cultcache(key = 15)]
pub detail: String,
#[cultcache(key = 16)]
pub updated_at: String,
#[cultcache(key = 17, default)]
pub image_mean_rgb: Vec<u32>,
#[cultcache(key = 18, default)]
pub image_peak_rgb: Vec<u32>,
#[cultcache(key = 19, default)]
pub color_evidence_move_ids: Vec<String>,
#[cultcache(key = 20, default)]
pub color_evidence_pixel_counts: Vec<u32>,
#[cultcache(key = 21, default)]
pub rejected_stale_count: u64,
#[cultcache(key = 22, default)]
pub rejected_radius_count: u64,
#[cultcache(key = 23, default)]
pub rejected_bounds_count: u64,
#[cultcache(key = 24, default)]
pub rejected_continuity_count: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "muninn.move_evidence_transport_health", schema = "muninn.move_evidence_transport_health.v1")]
#[cultcache(
type = "muninn.move_evidence_transport_health",
schema = "muninn.move_evidence_transport_health.v1"
)]
pub struct MuninnMoveEvidenceTransportHealthRecord {
#[cultcache(key = 0)] pub health_id: String,
#[cultcache(key = 1)] pub host_id: String,
#[cultcache(key = 2)] pub stream_id: String,
#[cultcache(key = 3)] pub produced_frames: u64,
#[cultcache(key = 4)] pub local_ring_admissions: u64,
#[cultcache(key = 5)] pub remote_handoffs: u64,
#[cultcache(key = 6)] pub remote_sends: u64,
#[cultcache(key = 7)] pub updated_at: String,
#[cultcache(key = 0)]
pub health_id: String,
#[cultcache(key = 1)]
pub host_id: String,
#[cultcache(key = 2)]
pub stream_id: String,
#[cultcache(key = 3)]
pub produced_frames: u64,
#[cultcache(key = 4)]
pub local_ring_admissions: u64,
#[cultcache(key = 5)]
pub remote_handoffs: u64,
#[cultcache(key = 6)]
pub remote_sends: u64,
#[cultcache(key = 7)]
pub updated_at: String,
}
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
@ -1421,10 +1515,12 @@ cultmesh_rs::cultmesh_documents!(OdinDocuments {
VoidBotSwarmStateSnapshotCompatRecord => VOIDBOT_SWARM_STATE_SNAPSHOT_SCHEMA,
IdunnDesiredDaemonRecord => IDUNN_DESIRED_DAEMON_SCHEMA,
IdunnDaemonHealthRecord => IDUNN_DAEMON_HEALTH_SCHEMA,
IdunnSignedHealthAdmissionRecord => IDUNN_SIGNED_HEALTH_ADMISSION_SCHEMA,
IdunnKeepaliveDecisionRecord => IDUNN_KEEPALIVE_DECISION_SCHEMA,
IdunnRestartRequestRecord => IDUNN_RESTART_REQUEST_SCHEMA,
IdunnRestartResultRecord => IDUNN_RESTART_RESULT_SCHEMA,
IdunnDeploymentRequestRecord => IDUNN_DEPLOYMENT_REQUEST_SCHEMA,
IdunnCurrentDeploymentRequestRecord => IDUNN_CURRENT_DEPLOYMENT_REQUEST_SCHEMA,
IdunnDeploymentResultRecord => IDUNN_DEPLOYMENT_RESULT_SCHEMA,
IdunnLifecycleCommandRecord => IDUNN_LIFECYCLE_COMMAND_SCHEMA,
IdunnReleaseTargetRecord => IDUNN_RELEASE_TARGET_SCHEMA,

6
crates/odin-core/src/idunn.rs
View file

@ -257,11 +257,7 @@ mod tests {
fn warming_daemon_is_observed_without_actuation_or_promotion() {
let mut desired = desired(Some("systemctl restart epiphany"));
desired.deploy_command = Some("deploy epiphany".to_string());
let plan = plan_keepalive(
&desired,
&health("warming"),
"2026-07-17T00:00:02Z",
);
let plan = plan_keepalive(&desired, &health("warming"), "2026-07-17T00:00:02Z");
assert_eq!(plan.decision.action, "observe");
assert!(plan.decision.reason.contains("provider-owned progress"));

15
docs/idunn.md
View file

@ -110,8 +110,10 @@ The current typed records are:
```text
idunn.desired_daemon.v1
idunn.daemon_health.v1
idunn.signed_health_admission.v1
idunn.keepalive_decision.v1
idunn.deployment_request.v1
idunn.deployment_request.v2
idunn.current_deployment_request.v1
idunn.deployment_result.v1
idunn.release_target.v1
idunn.deployment_artifact.v1
@ -184,9 +186,12 @@ idunn.rudp_health_ingress.v1
and then publishes `idunn.rudp_health_ingress.v1`. The Starfire local
supervisor binds `0.0.0.0:17870` explicitly so host-local and WireGuard
publishers can use the configured Idunn health endpoint. That ingress
accepts only raw `idunn.daemon_health` CultNet document puts on the `schema`
channel, decodes the typed MessagePack payload, and writes it into the
keepalive store. Each one-shot publisher gets its own RUDP session from its
accepts raw `idunn.daemon_health` CultNet document puts and the dedicated
signed Epiphany runtime-health schema on the `schema` channel. Generic health
is observation-only. Epiphany promotion requires the separately persisted
`idunn.signed_health_admission.v1`, an exact latest
`idunn.deployment_request.v2`, a pinned host signature, current matching
health, and a fresh active admission. Each one-shot publisher gets its own RUDP session from its
UDP source address and the session is discarded after a delivered health
frame. Windows UDP `ConnectionReset`/`ConnectionAborted` reports from closed
one-shot clients are nonfatal ingress noise, not a reason to kill the worker.
@ -285,7 +290,7 @@ idunn.rudp_health_ingress.v1
execution. Operators publish `idunn.lifecycle_command.v1` with `idunn restart
--daemon <id>` or `idunn redeploy --daemon <id>`. The running supervisor
claims pending records for its target catalog, writes the corresponding
`idunn.restart_request.v1` or `idunn.deployment_request.v1`, executes through
`idunn.restart_request.v1` or `idunn.deployment_request.v2`, executes through
the same guarded actuator path used by health-driven decisions, and records
the matching result document. If a target lacks restart or deploy authority,
Idunn marks the lifecycle command rejected instead of inventing authority.

4
scripts/linux/idunn-yggdrasil.service
View file

@ -8,7 +8,9 @@ Requires=docker.service
Type=simple
User=idunn
Group=idunn
ExecStart=/usr/local/bin/idunn --swarm-profile yggdrasil-local --repo-root /srv/odin/source --store /var/lib/gamecult/idunn/yggdrasil.keepalive.cc --release-authority-store /srv/bifrost/state/repository-release-authority.cc --rudp-health-bind 10.77.0.1:17870 --command-timeout-seconds 3600 --execute
# Operational cancellation ceiling only. Target daemons own readiness through
# typed health; this duration must never be interpreted as deployment success.
ExecStart=/usr/local/bin/idunn --swarm-profile yggdrasil-local --repo-root /srv/odin/source --store /var/lib/gamecult/idunn/yggdrasil.keepalive.cc --release-authority-store /srv/bifrost/state/repository-release-authority.cc --rudp-health-bind 10.77.0.1:17870 --trusted-epiphany-health-identity-store /etc/gamecult/idunn/epiphany-health-identity.ccmp --command-timeout-seconds 21600 --execute
Restart=always
RestartSec=5
TimeoutStopSec=20

9
src/Idunn/README.md
View file

@ -186,8 +186,10 @@ external-owned.
- `idunn.desired_daemon.v1`
- `idunn.daemon_health.v1`
- `idunn.signed_health_admission.v1`
- `idunn.keepalive_decision.v1`
- `idunn.deployment_request.v1`
- `idunn.deployment_request.v2`
- `idunn.current_deployment_request.v1`
- `idunn.deployment_result.v1`
- `idunn.release_target.v1`
- `idunn.deployment_artifact.v1`
@ -244,6 +246,11 @@ publication, not restart/deploy authority. On Windows, UDP reset reports from
closed one-shot publishers are treated as nonfatal ingress noise so one accepted
health frame cannot kill the listener.
Epiphany uses a separate signed runtime-health schema. Idunn joins that document
to the exact latest Bifrost-authorized deployment request and persists
`idunn.signed_health_admission.v1`. The privileged admission verifier reads only
that fresh, current, request-bound record; generic health remains observation.
When a fresh daemon-published RUDP health record exists for a target, Idunn uses
that record for keepalive planning. The record must match the daemon id, health
contract, RUDP transport, and freshness window. If it is absent or stale, Idunn

3
vendor/cultcache-rs/Cargo.toml vendored
View file

@ -22,6 +22,9 @@ serde_bytes = "0.11"
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
[target.'cfg(windows)'.dependencies]
windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] }
[dev-dependencies]
pretty_assertions = "1.4"
tempfile = "3"

117
vendor/cultcache-rs/src/lib.rs vendored
View file

@ -8,6 +8,7 @@ use std::collections::BTreeSet;
use std::fs;
use std::fs::File;
use std::fs::OpenOptions;
use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
@ -41,7 +42,7 @@ macro_rules! cultcache_registry {
};
}
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultCacheEnvelope {
pub key: String,
@ -54,6 +55,13 @@ pub struct CultCacheEnvelope {
pub schema_id: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CultCacheExpectedEnvelope {
pub key: String,
pub r#type: String,
pub current: Option<CultCacheEnvelope>,
}
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
struct PersistedStoreSnapshot(
String,
@ -128,6 +136,49 @@ impl SingleFileMessagePackBackingStore {
&self.path
}
pub fn compare_exchange(
&self,
expected: &[CultCacheExpectedEnvelope],
replacements: &[CultCacheEnvelope],
) -> Result<bool> {
self.with_exclusive_lock(|| {
let mut entries = self.read_all_unlocked()?;
for condition in expected {
let current = entries
.iter()
.find(|entry| entry.r#type == condition.r#type && entry.key == condition.key);
if current != condition.current.as_ref() {
return Ok(false);
}
if let Some(envelope) = condition.current.as_ref()
&& (envelope.r#type != condition.r#type || envelope.key != condition.key)
{
return Err(anyhow!(
"CultCache compare-exchange expectation identity differs from its envelope"
));
}
}
let mut replacement_ids = BTreeSet::new();
for replacement in replacements {
if replacement.key.trim().is_empty() || replacement.r#type.trim().is_empty() {
return Err(anyhow!(
"CultCache compare-exchange replacement identity is empty"
));
}
if !replacement_ids.insert(entry_id(replacement)) {
return Err(anyhow!(
"CultCache compare-exchange contains a duplicate replacement"
));
}
}
entries.retain(|entry| !replacement_ids.contains(&entry_id(entry)));
entries.extend_from_slice(replacements);
entries.sort_by_key(entry_id);
self.write_all_unlocked(&entries)?;
Ok(true)
})
}
fn read_all_unlocked(&self) -> Result<Vec<CultCacheEnvelope>> {
if !self.path.exists() {
return Ok(Vec::new());
@ -150,19 +201,25 @@ impl SingleFileMessagePackBackingStore {
let bytes = rmp_serde::to_vec(&encode_store_snapshot(entries))
.context("failed to encode MessagePack")?;
let tmp_path = temporary_path_for(&self.path);
fs::write(&tmp_path, bytes)
let mut temporary = OpenOptions::new()
.create_new(true)
.write(true)
.open(&tmp_path)
.with_context(|| format!("failed to create {}", tmp_path.display()))?;
temporary
.write_all(&bytes)
.with_context(|| format!("failed to write {}", tmp_path.display()))?;
if self.path.exists() {
fs::remove_file(&self.path)
.with_context(|| format!("failed to replace {}", self.path.display()))?;
temporary
.sync_all()
.with_context(|| format!("failed to sync {}", tmp_path.display()))?;
drop(temporary);
replace_file_atomically(&tmp_path, &self.path)?;
#[cfg(unix)]
if let Some(parent) = self.path.parent() {
File::open(parent)
.and_then(|directory| directory.sync_all())
.with_context(|| format!("failed to sync {}", parent.display()))?;
}
fs::rename(&tmp_path, &self.path).with_context(|| {
format!(
"failed to move {} to {}",
tmp_path.display(),
self.path.display()
)
})?;
Ok(())
}
@ -212,6 +269,42 @@ impl SingleFileMessagePackBackingStore {
}
}
#[cfg(unix)]
fn replace_file_atomically(source: &Path, destination: &Path) -> Result<()> {
fs::rename(source, destination).with_context(|| {
format!(
"failed to atomically replace {} with {}",
destination.display(),
source.display()
)
})
}
#[cfg(windows)]
fn replace_file_atomically(source: &Path, destination: &Path) -> Result<()> {
use std::os::windows::ffi::OsStrExt;
use windows_sys::Win32::Storage::FileSystem::{
MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, MoveFileExW,
};
let source: Vec<u16> = source.as_os_str().encode_wide().chain(Some(0)).collect();
let destination: Vec<u16> = destination
.as_os_str()
.encode_wide()
.chain(Some(0))
.collect();
let result = unsafe {
MoveFileExW(
source.as_ptr(),
destination.as_ptr(),
MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
)
};
if result == 0 {
return Err(std::io::Error::last_os_error()).context("atomic CultCache replacement failed");
}
Ok(())
}
impl CacheBackingStore for SingleFileMessagePackBackingStore {
fn pull_all(&self) -> Result<Vec<CultCacheEnvelope>> {
self.with_shared_lock(|| self.read_all_unlocked())