Skip to main content

max / makenotwork

synckit: give Idempotency runtime teeth, unify rotation decrypt (audit Phase 4) Architecture remediation from the 2026-07-02 /audit arbiter: - Idempotency is no longer a runtime no-op. An `Unsafe` variant plus may_retry() gates the retry loop, so a non-idempotent operation is attempted exactly once by construction rather than by the caller remembering not to wrap it. create_subscription_checkout is marked Unsafe (a blind retry could mint a second Stripe Checkout session). Both `let _ = idempotency` discards removed; an integration test pins exactly-once behaviour on a transient 503. - The live rotation pull path reimplemented a fallback-less key selection while the tested selection (with the unknown-key_id primary->pending fallback) sat behind #[allow(dead_code)]. Both are replaced by one generic decrypt_with_rotation_keys used by the live path and the tests, so the code that runs in production is the code the tests exercise. Removes the dead master_key_id read; PullChangeEntry gains Clone for the fallback. Gate: 406 crate tests pass, cargo clippy --all-targets clean, GoingsOn cargo check clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-03 01:45 UTC
Commit: 2ed0884c9b1bd1ce21053d74e16a9e92fb80beee
Parent: cfd9de7
5 files changed, +93 insertions, -65 deletions
M shared/synckit-client/src/client/helpers.rs +50 -56
@@ -13,9 +13,10 @@ use super::{BASE_DELAY, MAX_RETRIES, SyncKitClient};
13 13 /// Retry replays the request, so it is only safe when the server treats the
14 14 /// operation idempotently. Requiring this argument makes that judgement an
15 15 /// explicit, reviewable decision: a new endpoint cannot be fed to a retry
16 - /// helper without naming why replay is harmless here. There is deliberately no
17 - /// `Unsafe`/`Never` variant — an operation that cannot tolerate replay simply
18 - /// must not call a retry helper.
16 + /// helper without naming why replay is harmless here — and the judgement now has
17 + /// runtime teeth, not just documentary value: [`may_retry`](Self::may_retry)
18 + /// gates the retry loop, so an [`Unsafe`](Self::Unsafe) operation physically
19 + /// cannot be auto-replayed.
19 20 #[derive(Clone, Copy, Debug)]
20 21 pub(super) enum Idempotency {
21 22 /// Read-only, or a write whose replay yields the same end state (a `GET`,
@@ -24,6 +25,17 @@ pub(super) enum Idempotency {
24 25 /// Carries a client-generated idempotency key (e.g. push's `batch_id`) so
25 26 /// the server collapses duplicate deliveries into one effect.
26 27 Keyed,
28 + /// Not idempotent and not idempotency-keyed: replaying it would mint
29 + /// duplicate external state (e.g. a second Stripe Checkout session). The
30 + /// retry helpers attempt it exactly once.
31 + Unsafe,
32 + }
33 +
34 + impl Idempotency {
35 + /// Whether the retry helpers may replay this operation on a transient error.
36 + fn may_retry(self) -> bool {
37 + matches!(self, Idempotency::Safe | Idempotency::Keyed)
38 + }
27 39 }
28 40
29 41 /// Current on-wire HLC-envelope version, written into the `__skver` tag.
@@ -69,10 +81,11 @@ impl SyncKitClient {
69 81 F: FnMut() -> Fut,
70 82 Fut: std::future::Future<Output = Result<reqwest::Response>>,
71 83 {
72 - let _ = idempotency;
84 + // A non-idempotent operation gets exactly one attempt: zero retries.
85 + let max_attempts = if idempotency.may_retry() { MAX_RETRIES } else { 0 };
73 86 let mut last_err = None;
74 87
75 - for attempt in 0..=MAX_RETRIES {
88 + for attempt in 0..=max_attempts {
76 89 match operation().await {
77 90 Ok(resp) => return Ok(resp),
78 91 Err(err) => {
@@ -80,7 +93,7 @@ impl SyncKitClient {
80 93 return Err(err);
81 94 }
82 95
83 - if attempt < MAX_RETRIES {
96 + if attempt < max_attempts {
84 97 let delay = retry_delay(&err, attempt);
85 98 tracing::debug!(
86 99 attempt = attempt + 1,
@@ -116,17 +129,18 @@ impl SyncKitClient {
116 129 Fut: std::future::Future<Output = Result<reqwest::Response>>,
117 130 T: serde::de::DeserializeOwned,
118 131 {
119 - let _ = idempotency;
132 + // A non-idempotent operation gets exactly one attempt: zero retries.
133 + let max_attempts = if idempotency.may_retry() { MAX_RETRIES } else { 0 };
120 134 let mut last_err = None;
121 135
122 - for attempt in 0..=MAX_RETRIES {
136 + for attempt in 0..=max_attempts {
123 137 match operation().await {
124 138 Ok(resp) => {
125 139 match resp.json::<T>().await {
126 140 Ok(parsed) => return Ok(parsed),
127 141 Err(e) => {
128 142 let err = SyncKitError::Http(e);
129 - if attempt < MAX_RETRIES {
143 + if attempt < max_attempts {
130 144 let delay = retry_delay(&err, attempt);
131 145 tracing::debug!(
132 146 attempt = attempt + 1,
@@ -146,11 +160,11 @@ impl SyncKitClient {
146 160 return Err(err);
147 161 }
148 162
149 - if attempt < MAX_RETRIES {
163 + if attempt < max_attempts {
150 164 let delay = retry_delay(&err, attempt);
151 165 tracing::debug!(
152 166 attempt = attempt + 1,
153 - max_retries = MAX_RETRIES,
167 + max_retries = max_attempts,
154 168 delay_ms = delay.as_millis() as u64,
155 169 error = %err,
156 170 "Transient error, retrying after backoff",
@@ -295,55 +309,35 @@ impl SyncKitClient {
295 309 })
296 310 }
297 311
298 - /// Decrypt with key selection based on key_id. Used during rotation when
299 - /// pulled entries may be encrypted with different keys.
300 - #[allow(dead_code)]
301 - pub(super) fn decrypt_change_multi_key(
312 + /// Decrypt a pulled entry during a rotation window, selecting the key by the
313 + /// entry's `key_id`. Generic over the decrypt step so both the plain-pull
314 + /// (`ChangeEntry`) and rich-pull (`PulledChange`) paths share exactly this
315 + /// selection logic — including the unknown-`key_id` fallback that tries the
316 + /// primary key then the pending key. Previously the live pull path
317 + /// reimplemented a fallback-less variant while the tested one sat unused.
318 + pub(super) fn decrypt_with_rotation_keys<T, F>(
302 319 entry: PullChangeEntry,
303 320 primary_key: &[u8; 32],
304 321 primary_key_id: i32,
305 322 pending_key: &[u8; 32],
306 323 pending_key_id: i32,
307 - ) -> Result<ChangeEntry> {
324 + decrypt_fn: &F,
325 + ) -> Result<T>
326 + where
327 + F: Fn(PullChangeEntry, &[u8; 32]) -> Result<T>,
328 + {
308 329 let effective_key_id = entry.key_id.unwrap_or(1);
309 - let key = if effective_key_id == pending_key_id {
310 - pending_key
330 + if effective_key_id == pending_key_id {
331 + decrypt_fn(entry, pending_key)
311 332 } else if effective_key_id == primary_key_id || effective_key_id <= 1 {
312 - primary_key
333 + decrypt_fn(entry, primary_key)
313 334 } else {
314 - // Unknown key_id — try primary, then pending
315 - match Self::decrypt_change_with_key_inner(&entry, primary_key) {
316 - Ok(result) => return Ok(result),
317 - Err(_) => pending_key,
318 - }
319 - };
320 - Self::decrypt_change_with_key(entry, key)
321 - }
322 -
323 - /// Inner decryption that borrows the entry (for fallback logic).
324 - #[allow(dead_code)]
325 - fn decrypt_change_with_key_inner(entry: &PullChangeEntry, master_key: &[u8; 32]) -> Result<ChangeEntry> {
326 - let ctx = crypto::AeadContext::entry(&entry.table, &entry.row_id);
327 - let (hlc, data) = match entry.data {
328 - Some(ref value) => {
329 - let decrypted = crypto::decrypt_json_aad(value, master_key, &ctx)?;
330 - Self::split_hlc_envelope(decrypted, entry.device_id.as_uuid(), entry.timestamp.timestamp_millis())?
335 + // Unknown key_id — try primary, then fall back to pending.
336 + match decrypt_fn(entry.clone(), primary_key) {
337 + Ok(result) => Ok(result),
338 + Err(_) => decrypt_fn(entry, pending_key),
331 339 }
332 - None => (
333 - Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id.as_uuid()),
334 - None,
335 - ),
336 - };
337 -
338 - Ok(ChangeEntry {
339 - table: entry.table.clone(),
340 - op: entry.op,
341 - row_id: entry.row_id.clone(),
342 - hlc,
343 - timestamp: entry.timestamp,
344 - data,
345 - extra: Default::default(),
346 - })
340 + }
347 341 }
348 342
349 343 /// Decrypt with a pre-loaded key. Used by `pull()` to avoid per-entry lock acquisition.
@@ -1084,7 +1078,7 @@ mod tests {
1084 1078 let plaintext = serde_json::json!({"v": "pending-payload"});
1085 1079 let entry = pull_entry_with(encrypt_with(&pending, plaintext.clone()), Some(7));
1086 1080
1087 - let decrypted = SyncKitClient::decrypt_change_multi_key(entry, &primary, 1, &pending, 7).unwrap();
1081 + let decrypted = SyncKitClient::decrypt_with_rotation_keys(entry, &primary, 1, &pending, 7, &SyncKitClient::decrypt_change_with_key).unwrap();
1088 1082 assert_eq!(decrypted.data.unwrap(), plaintext);
1089 1083 }
1090 1084
@@ -1095,7 +1089,7 @@ mod tests {
1095 1089 let plaintext = serde_json::json!({"v": "primary-payload"});
1096 1090 let entry = pull_entry_with(encrypt_with(&primary, plaintext.clone()), Some(3));
1097 1091
1098 - let decrypted = SyncKitClient::decrypt_change_multi_key(entry, &primary, 3, &pending, 9).unwrap();
1092 + let decrypted = SyncKitClient::decrypt_with_rotation_keys(entry, &primary, 3, &pending, 9, &SyncKitClient::decrypt_change_with_key).unwrap();
1099 1093 assert_eq!(decrypted.data.unwrap(), plaintext);
1100 1094 }
1101 1095
@@ -1111,7 +1105,7 @@ mod tests {
1111 1105 let plaintext = serde_json::json!({"v": "legacy-no-key-id"});
1112 1106 let entry = pull_entry_with(encrypt_with(&primary, plaintext.clone()), None);
1113 1107
1114 - let decrypted = SyncKitClient::decrypt_change_multi_key(entry, &primary, 5, &pending, 6).unwrap();
1108 + let decrypted = SyncKitClient::decrypt_with_rotation_keys(entry, &primary, 5, &pending, 6, &SyncKitClient::decrypt_change_with_key).unwrap();
1115 1109 assert_eq!(decrypted.data.unwrap(), plaintext);
1116 1110 }
1117 1111
@@ -1125,7 +1119,7 @@ mod tests {
1125 1119 let plaintext = serde_json::json!({"v": "via-fallback-primary"});
1126 1120 let entry = pull_entry_with(encrypt_with(&primary, plaintext.clone()), Some(42));
1127 1121
1128 - let decrypted = SyncKitClient::decrypt_change_multi_key(entry, &primary, 5, &pending, 6).unwrap();
1122 + let decrypted = SyncKitClient::decrypt_with_rotation_keys(entry, &primary, 5, &pending, 6, &SyncKitClient::decrypt_change_with_key).unwrap();
1129 1123 assert_eq!(decrypted.data.unwrap(), plaintext);
1130 1124 }
1131 1125
@@ -1138,7 +1132,7 @@ mod tests {
1138 1132 let plaintext = serde_json::json!({"v": "via-fallback-pending"});
1139 1133 let entry = pull_entry_with(encrypt_with(&pending, plaintext.clone()), Some(42));
1140 1134
1141 - let decrypted = SyncKitClient::decrypt_change_multi_key(entry, &primary, 5, &pending, 6).unwrap();
1135 + let decrypted = SyncKitClient::decrypt_with_rotation_keys(entry, &primary, 5, &pending, 6, &SyncKitClient::decrypt_change_with_key).unwrap();
1142 1136 assert_eq!(decrypted.data.unwrap(), plaintext);
1143 1137 }
1144 1138 }
M shared/synckit-client/src/client/subscription.rs +3 -1
@@ -308,7 +308,9 @@ impl SyncKitClient {
308 308 cap_bytes,
309 309 interval: interval.as_str(),
310 310 })?);
311 - self.retry_request_json(Idempotency::Safe, || {
311 + // Not idempotent: a blind auto-retry could mint a second Stripe Checkout
312 + // session. Attempt once; the user re-initiates on a transient failure.
313 + self.retry_request_json(Idempotency::Unsafe, || {
312 314 let req = self
313 315 .http
314 316 .post(&self.endpoints.subscription_checkout)
M shared/synckit-client/src/client/sync.rs +9 -7
@@ -224,17 +224,19 @@ impl SyncKitClient {
224 224
225 225 let changes = if has_pending {
226 226 let pending = pending_guard.as_ref().unwrap();
227 - let _primary_key_id = *self.master_key_id.read();
227 + let primary_key_id = *self.master_key_id.read();
228 228 pull_resp
229 229 .changes
230 230 .into_iter()
231 231 .map(|c| {
232 - let effective_key_id = c.key_id.unwrap_or(1);
233 - if effective_key_id == pending.key_id {
234 - decrypt_fn(c, &pending.key)
235 - } else {
236 - decrypt_fn(c, master_key)
237 - }
232 + Self::decrypt_with_rotation_keys(
233 + c,
234 + master_key,
235 + primary_key_id,
236 + &pending.key,
237 + pending.key_id,
238 + &decrypt_fn,
239 + )
238 240 })
239 241 .collect::<Result<Vec<_>>>()?
240 242 } else {
M shared/synckit-client/src/types.rs +1 -1
@@ -268,7 +268,7 @@ pub(crate) struct PullResponse {
268 268 /// `seq` and `device_id` are present in the server response and parsed for
269 269 /// completeness, but not used by the SDK — consumers track cursors, not
270 270 /// individual sequence numbers.
271 - #[derive(Deserialize)]
271 + #[derive(Deserialize, Clone)]
272 272 pub(crate) struct PullChangeEntry {
273 273 #[allow(dead_code)]
274 274 pub seq: i64,
M shared/synckit-client/tests/integration.rs +30
@@ -2072,6 +2072,36 @@ async fn retry_exhausts_all_attempts_on_persistent_503() {
2072 2072 }
2073 2073
2074 2074 #[tokio::test]
2075 + async fn unsafe_op_makes_exactly_one_attempt_on_transient_error() {
2076 + // create_subscription_checkout is Idempotency::Unsafe (a retry could mint a
2077 + // second Stripe session), so a transient 503 must NOT be retried.
2078 + let server = MockServer::start().await;
2079 +
2080 + Mock::given(method("POST"))
2081 + .and(path("/api/v1/sync/subscription/checkout"))
2082 + .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
2083 + .mount(&server)
2084 + .await;
2085 +
2086 + let client = authed_client(&server);
2087 + let err = client
2088 + .create_subscription_checkout(1_000_000_000, synckit_client::BillingInterval::Monthly)
2089 + .await
2090 + .unwrap_err();
2091 + assert!(matches!(err, SyncKitError::Server { status: 503, .. }));
2092 +
2093 + let requests = server.received_requests().await.unwrap();
2094 + let checkout_requests = requests
2095 + .iter()
2096 + .filter(|r| r.url.path() == "/api/v1/sync/subscription/checkout")
2097 + .count();
2098 + assert_eq!(
2099 + checkout_requests, 1,
2100 + "Unsafe op must be attempted exactly once, got {checkout_requests}"
2101 + );
2102 + }
2103 +
2104 + #[tokio::test]
2075 2105 async fn retry_not_attempted_on_404() {
2076 2106 let server = MockServer::start().await;
2077 2107