chore: remove global epoxy contention

Author: NathanFlurry

.agent/specs/epoxy-mutable-keys.md

@@ -0,0 +1,26 @@
+# Epoxy PR Fixes
+
+TODO items for the `04-04-chore_remove_global_epoxy_contention` PR.
+
+## Items
+
+- [ ] **Use BARE for `KvAcceptedValue` serialization.** `engine/packages/epoxy/src/keys/keys.rs:13` uses raw `serde_bare` with a comment claiming versioned protocol is unnecessary because accepted state is transient. We should still use the versioned BARE protocol path for consistency. Remove the comment justifying raw serde_bare and switch to the protocol-based serialization.
+
+- [ ] **Add doc comment to `LegacyCommittedValueKey`.** `engine/packages/epoxy/src/keys/keys.rs:73` has no doc comment. Add a brief comment explaining this is the v1 committed value key used for dual-read fallback, and link to the "Legacy Fallback" section in `engine/packages/epoxy/README.md`.
+
+- [ ] **Add file-level doc comment to `keys/keys.rs`.** Add a module-level comment at the top of `engine/packages/epoxy/src/keys/keys.rs` linking to `engine/packages/epoxy/README.md` for the full key layout and subspace structure.
+
+- [ ] **Delete `read_value.rs`, inline the read logic into each caller.** `get_local.rs` needs only the value (no cache), `get_optimistic.rs` needs value + cache. Inline the legacy fallback transaction directly into each op and remove `engine/packages/epoxy/src/ops/kv/read_value.rs` and its `mod` entry. Update `engine/CLAUDE.md` references to `read_value.rs`.
+
+- [ ] **Flatten `Proposal`/`Command`/`CommandKind` into a single key+value input.** Every caller wraps exactly one command in `vec![Command { kind: ... }]` and `from_proposal` bails if `len() != 1`. Remove the `Proposal`, `Command`, `CommandKind`, and `NoopCommand` types. `Input` should take key + value directly (or just `SetIfAbsentProposal`). Update callers in `pegboard/workflows/actor/keys.rs`, `pegboard/workflows/actor2/keys.rs`, `api-peer/src/internal.rs`, and test utils.
+
+- [ ] **Design request-level caching/mutability control for epoxy.** This is a design task -- write a spec in `.agent/specs/` before implementing. Key changes needed:
+  - **Versioned values:** Add version numbers or timestamps to committed values so ballot selection and commit handlers can distinguish newer writes from older ones (currently short-circuit on first committed value at `ballot.rs:126`, `commit.rs:15`, `accept.rs:29`).
+  - **Ballot selection:** Can't short-circuit on first committed value. Must allow new proposals for already-committed keys.
+  - **Changelog catch-up:** `changelog.rs:123-128` asserts duplicate entries have identical values (`ensure!(existing_value == entry.value)`). With mutable keys, same key can appear multiple times with different values. Fix: overwrite with newer value instead of asserting equality (entries are versionstamp-ordered).
+  - **Cache invalidation:** Restore the old purger workflow mechanism (deleted in this PR as `workflows/purger.rs`). No TTLs.
+  - **No quorum reads needed.** Caller controls staleness tradeoff at request level by choosing whether to use cache or do fresh fanout. Per-request `CachingBehavior` on reads (e.g. `Optimistic` = use cache, `SkipCache` = fresh fanout).
+
+- [ ] **Reword quorum read comment in `get_optimistic.rs:45`.** Replace "We cannot use quorum reads for the fanout read because the optimistic path is intentionally a best-effort lookup" with something like "The fanout asks any single remote replica rather than waiting for a quorum. Since committed values are immutable, any replica that has the value is authoritative. The tradeoff is that if all replicas holding the value are offline, the fanout returns None."
+
+- [ ] **Rename `EPOXY` to `EPOXY_V1` in `universaldb/src/utils/keys.rs`.** The constant currently named `EPOXY` is the v1 subspace integer. Rename it to `EPOXY_V1` (keeping the same integer value) for clarity now that `EPOXY_V2` exists. Update all references (e.g. `keys::legacy_subspace` in `engine/packages/epoxy/src/keys/mod.rs`).

.agent/todo/epoxy-pr-fixes.md

@@ -1,6 +1,6 @@
 # Actor Key Reservation
 
-Epoxy is used to globally store which datacenter an actor lives in for a given actor ID.
+Epoxy uses per-key Paxos to globally store which datacenter an actor key resolves to.
 
 ## Schema
 
@@ -12,8 +12,8 @@ The reservation ID includes the datacenter of where the actor lives.
 
 Actors have 2 touch points with Epoxy:
 
-- Reserve actor key (epaxos fast path = 1 RTT, epaxos slow path = 2 RTT)
-- Resolve actor ID for key (fast path = 0 RTT, slow path = 1 RTT to nearest DC that has the key)
+- Reserve actor key. Fresh keys use the per-key Paxos fast path in 1 RTT. If contention or recovery forces `Prepare`, the retry path is `Prepare -> PreAccept -> Commit`, which adds another round and makes a failed-then-retried write 3 RTT total.
+- Resolve actor ID for key. Hot reads are local. Cache misses use one best-effort fanout round to the nearest datacenter that has the key.
 
 Resolving actor ID uses `kv_get_optimistic`, which assumes a value does not change after being set. This allows us to cache the actor's datacenter locally and never have to read from other nodes once we've resolved it once.
 
@@ -98,4 +98,3 @@ _This section needs more elaboration on edge cases._
 ## Current Limitation: Keys are tied to a datacenter
 
 Because we cannot change the value of the reservation ID, actors can only be created in the original for a given ID.
-

Cargo.lock

@@ -26,3 +26,25 @@ When changing a versioned VBARE schema, follow the existing migration pattern.
      - `engine/sdks/rust/runner-protocol/src/lib.rs` `PROTOCOL_MK2_VERSION`
      - `engine/sdks/typescript/runner/src/mod.ts` `PROTOCOL_VERSION`
    - Update the Rust latest re-export in `engine/sdks/rust/runner-protocol/src/lib.rs` to the new generated module.
+
+## Epoxy durable keys
+
+- Keep shared epoxy tuple layouts in `engine/packages/epoxy/src/keys/keys.rs`. Replica modules should import those `FormalKey` types instead of redefining local tuple paths.
+- When adding a new epoxy tuple segment, add the shared constant in `engine/packages/universaldb/src/utils/keys.rs` first so all helpers serialize the same on-disk path.
+- In epoxy's per-key replica handlers, scope transactions with `keys::subspace(replica_id)` and use the shared `KvValueKey`, `KvBallotKey`, and `KvAcceptedKey` types directly. Successful Prepare replies should echo the promised ballot in `PrepareResponseOk.highest_ballot`, while any prior accepted record comes from `kv/{key}/accepted`.
+- Epoxy changelog writes use versionstamped keys. `Transaction::write` only does plain `pack`, so changelog code should build the full key with `keys::subspace(replica_id).pack_with_versionstamp(...)`, substitute the incomplete versionstamp, and then write it with `tx.set`.
+- Epoxy changelog catch-up must stick to one active source replica for the entire pagination loop. Changelog cursors are per-replica versionstamps, so switching sources mid-stream is invalid.
+- During epoxy replica bootstrap, `catch_up_replica` should treat "no active source replica yet" as a successful no-op. Fresh clusters can start with every replica in `Joining`, so retrying forever on a missing active source stalls initial bring-up.
+- Epoxy replica data lives under per-replica subspaces inside the shared UniversalDB. If a workflow needs another replica's changelog or KV state for coordination or metrics, read that replica's `keys::subspace(replica_id)` directly instead of adding a second RPC just to count local durable state.
+- Epoxy quorum helpers use Fast Paxos math, not the older `f`-based EPaxos formula. Keep `slow_q = floor(n / 2) + 1`, derive `fast_q = n - floor((slow_q - 1) / 2)`, and keep sender-excluded fanout quorums as those sizes minus one for `Fast`, `Slow`, and `All` while `Any` still targets one remote response.
+- Broadcast the updated epoxy config to all replicas before sending `BeginLearning`. Commit fanout uses `get_all_replicas(config)`, so the learner only receives live commits during catch-up after the rest of the cluster knows about it.
+- Epoxy v2 durable state lives under `keys::subspace(replica_id)`, which now points at the `EPOXY_V2` subspace. Legacy committed values remain read-only under `keys::legacy_subspace(replica_id)`.
+- Reuse the common local read order from `engine/packages/epoxy/src/ops/kv/read_value.rs`: v2 `KvValueKey`, legacy `LegacyCommittedValueKey`, legacy `KvValueKey`, then v2 `KvOptimisticCacheKey`.
+- Ballot selection should only read legacy committed values from `keys::legacy_subspace(replica_id)`. Do not write new ballot, accepted, config, or changelog state into the legacy subspace.
+- `engine/packages/epoxy/src/ops/propose.rs` keeps a compatibility `Proposal` and `Command` surface, but the v2 consensus path only supports a single-key set-if-absent value. Use `CheckAndSetCommand { expect_one_of: [None], new_value: Some(...) }` for actor key reservation, or a single `SetCommand` with `value: Some(...)` when you only need the same immutable insert semantics.
+- Epoxy slow-path Prepare retries should not immediately rebump ballots in a tight loop. Keep the retry path in `engine/packages/epoxy/src/ops/propose.rs` on a bounded exponential backoff with jitter: start at 10 ms, double each retry to a 1 s base cap, add 0-50% jitter, and fail after 10 retries instead of spinning until the overall request timeout.
+- The per-key Prepare and PreAccept rounds should include the local replica by sending a self-targeted epoxy request through `http_client::send_message`. Self-routing lands in `message_request` directly, so local and remote replicas share the same handler path and quorum accounting.
+- Keep the epoxy HTTP transport split between `/epoxy/message` for replica RPCs and `/epoxy/changelog-read` for pagination. Route handlers should reject the wrong request kind so changelog reads cannot silently drift back onto the generic message endpoint.
+- When you add fields to epoxy workflow state structs, mark the new fields with `#[serde(default)]`. Gasoline replays serialized workflow state, so new coordinator or replica state fields must deserialize cleanly from older runs.
+- Epoxy changelog GC should truncate through the latest currently visible changelog key, not clear the whole subspace blindly. Use `KeySelector::last_less_than(changelog_subspace_end)` plus `end_of_key_range(last_key)` so concurrent later appends stay outside the cleared range.
+- Epoxy integration tests that spin up `tests/common::TestCtx` must call `shutdown()` before returning. Dropping the harness alone leaves workflow workers and API tasks alive long enough to interfere with later tests in the same process.

docs-internal/engine/ACTOR_KEY_RESERVATION.md

@@ -1,5 +1,8 @@
 use anyhow::*;
-use epoxy_protocol::protocol::{ReplicaId, SlotId};
+use epoxy::{
+	ops::propose::{Command, CommandKind, Proposal, SetCommand},
+	protocol::ReplicaId,
+};
 use gas::prelude::*;
 use indexmap::IndexMap;
 use rivet_api_builder::ApiCtx;
@@ -226,13 +229,11 @@ pub async fn get_epoxy_replica_debug(
 ) -> Result<GetEpoxyReplicaDebugResponse> {
 	let replica_id = ctx.config().epoxy_replica_id();
 
-	let (config, ballot, instance_number) = ctx
+	let config = ctx
 		.udb()?
 		.run(|tx| async move {
 			let config = epoxy::utils::read_config(&tx, replica_id).await?;
-			let ballot = epoxy::replica::ballot::get_ballot(&tx, replica_id).await?;
-			let instance_number = epoxy::utils::read_instance_number(&tx, replica_id).await?;
-			Result::Ok((config, ballot, instance_number))
+			Result::Ok(config)
 		})
 		.await?;
 
@@ -284,11 +285,12 @@ pub async fn get_epoxy_replica_debug(
 		},
 		state: EpoxyReplicaDebugState {
 			ballot: EpoxyBallot {
-				epoch: ballot.epoch,
-				ballot: ballot.ballot,
-				replica_id: ballot.replica_id,
+				epoch: 0,
+				ballot: 0,
+				replica_id,
 			},
-			instance_number,
+			// Epoxy v2 no longer maintains replica-global ballot or instance counters.
+			instance_number: 0,
 		},
 	})
 }
@@ -310,7 +312,7 @@ pub struct GetEpoxyKeyDebugResponse {
 #[derive(Serialize, Deserialize, Clone)]
 pub struct EpoxyKeyInstance {
 	pub replica_id: ReplicaId,
-	pub slot_id: SlotId,
+	pub slot_id: u64,
 	pub log_entry: Option<EpoxyKeyLogEntry>,
 }
 
@@ -325,7 +327,7 @@ pub struct EpoxyKeyLogEntry {
 #[derive(Serialize, Deserialize, Clone)]
 pub struct EpoxyInstance {
 	pub replica_id: ReplicaId,
-	pub slot_id: SlotId,
+	pub slot_id: u64,
 }
 
 /// Returns debug information for a specific key on this replica.
@@ -342,46 +344,33 @@ pub async fn get_epoxy_key_debug(
 	let key_bytes = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, &path.key)
 		.context("invalid base64 key")?;
 
-	let instances = ctx
-		.udb()?
-		.run(|tx| {
-			let key_bytes = key_bytes.clone();
-			async move {
-				let protocol_instances =
-					epoxy::utils::read_key_instances(&tx, replica_id, key_bytes).await?;
-
-				let mut instances = Vec::new();
-				for instance in protocol_instances {
-					let log_entry =
-						epoxy::utils::read_log_entry(&tx, replica_id, &instance).await?;
-					instances.push(EpoxyKeyInstance {
-						replica_id: instance.replica_id,
-						slot_id: instance.slot_id,
-						log_entry: log_entry.map(|entry| EpoxyKeyLogEntry {
-							status: format!("{:?}", entry.state),
-							ballot: EpoxyBallot {
-								epoch: entry.ballot.epoch,
-								ballot: entry.ballot.ballot,
-								replica_id: entry.ballot.replica_id,
-							},
-							seq: entry.seq,
-							deps: entry
-								.deps
-								.into_iter()
-								.map(|d| EpoxyInstance {
-									replica_id: d.replica_id,
-									slot_id: d.slot_id,
-								})
-								.collect(),
-						}),
-					});
-				}
-
-				Result::Ok(instances)
-			}
+	let local_value = ctx
+		.op(epoxy::ops::kv::get_local::Input {
+			replica_id,
+			key: key_bytes,
 		})
 		.await?;
 
+	let instances = local_value
+		.value
+		.map(|_| {
+			vec![EpoxyKeyInstance {
+				replica_id,
+				slot_id: 0,
+				log_entry: Some(EpoxyKeyLogEntry {
+					status: "committed".to_string(),
+					ballot: EpoxyBallot {
+						epoch: 0,
+						ballot: 0,
+						replica_id,
+					},
+					seq: 0,
+					deps: Vec::new(),
+				}),
+			}]
+		})
+		.unwrap_or_default();
+
 	// Compute instances by status
 	let mut instances_by_status = IndexMap::new();
 	for instance in &instances {
@@ -536,6 +525,7 @@ pub async fn get_epoxy_kv_optimistic(
 		.op(epoxy::ops::kv::get_optimistic::Input {
 			replica_id,
 			key: key_bytes,
+			caching_behavior: epoxy::protocol::CachingBehavior::Optimistic,
 		})
 		.await?;
 
@@ -567,7 +557,6 @@ pub async fn set_epoxy_kv(
 	body: SetEpoxyKvRequest,
 ) -> Result<SetEpoxyKvResponse> {
 	use base64::Engine;
-	use epoxy_protocol::protocol;
 
 	let key_bytes = base64::engine::general_purpose::STANDARD
 		.decode(&path.key)
@@ -582,16 +571,21 @@ pub async fn set_epoxy_kv(
 		})
 		.transpose()?;
 
+	let value_bytes = value_bytes.ok_or_else(|| {
+		anyhow!("epoxy v2 debug set only supports immutable set-if-absent writes")
+	})?;
+
 	let result = ctx
 		.op(epoxy::ops::propose::Input {
-			proposal: protocol::Proposal {
-				commands: vec![protocol::Command {
-					kind: protocol::CommandKind::SetCommand(protocol::SetCommand {
+			proposal: Proposal {
+				commands: vec![Command {
+					kind: CommandKind::SetCommand(SetCommand {
 						key: key_bytes,
-						value: value_bytes,
+						value: Some(value_bytes),
 					}),
 				}],
 			},
+			mutable: false,
 			purge_cache: true,
 			target_replicas: None,
 		})

engine/CLAUDE.md

@@ -14,6 +14,7 @@ gas.workspace = true
 epoxy-protocol.workspace = true
 futures-util.workspace = true
 lazy_static.workspace = true
+rand.workspace = true
 reqwest.workspace = true
 rivet-api-builder.workspace = true
 rivet-config.workspace = true
@@ -34,7 +35,6 @@ tracing.workspace = true
 universaldb.workspace = true
 url.workspace = true
 uuid.workspace = true
-vbare.workspace = true
 
 [dev-dependencies]
 gas.workspace = true