A scheduled job that needs 90 to 180 seconds to produce a single output file looks harmless until a deploy lands while it is still running. The deployment controller drains the old task and starts a replacement. For two or three minutes both replicas are alive, both read the same input snapshot and both intend to write the same logical output. Without idempotent output keying, they write it twice, and the two writes can differ. Any consumer that reads during that window can pick up state assembled from two different runs.
The race shows up anywhere a long-running task publishes to shared storage and the orchestrator uses rolling deployments, which covers most production batch pipelines. The failure is quiet. Nothing crashes, and the logs show two successful task completions. The corruption lives entirely in the output, and it surfaces later as a downstream decision made on data that never existed as a coherent snapshot.
Why Rolling Deployments Break Long-Running Tasks
It comes down to a mismatch between two time scales. A rolling deployment is designed around request handlers that finish in milliseconds, so a few seconds of overlap between old and new replicas is invisible. A task that runs for minutes doesn’t fit that assumption. When the controller starts the new replica, the old one is often most of the way through its work. It holds partial results in memory and is headed for the same destination key. The orchestrator considers both healthy. It only checks liveness.
Most teams reach first for at-least-once scheduling with a fixed output path. The task computes its result and writes to a known location; newest write wins. That model is fine with a single task. Under deployment overlap it becomes last-writer-wins on a destination that two writers reached through different code paths or different partial reads. If the new build changed how a field is aggregated, the surviving file depends on which replica finished last, which is nondeterministic.
You could serialize the writers instead, with a distributed lock: a lease in etcd or ZooKeeper that lets only one task write at a time. But that trades a correctness problem for an availability problem. A task that holds a lease for 3 minutes and then hits a stop-the-world pause or a network partition puts you in a bind. Either the lease expires and a second task proceeds, which is the duplication you wanted to prevent, or you hold the lease conservatively and a crashed task blocks all progress until an operator intervenes. Locks don’t remove the problem; they move it.
Detecting Divergent Writes Before They Reach Downstream Consumers
Duplicate writes are invisible by default. On a store that keeps only the latest object, the second write erases the evidence of the first. The first instrumentation step is to turn on object versioning for the output prefix, which costs storage but turns a silent overwrite into a history you can inspect.
With versioning on, a duplicate write is detectable as more than one version of the same key inside a single scheduled window. On its own, that isn’t a defect: an idempotent rewrite of identical bytes is benign. The real signal is divergence: two versions of the same logical output whose checksums differ. The scan below walks every version under a window prefix, groups by key, and reports only keys whose versions carry more than one distinct entity tag (ETag), proof that two runs produced different bytes for the same window. A caveat on the mechanism: S3 ETags are only a reliable content hash for single-part, non-KMS-encrypted objects. For multipart objects the ETag depends on part boundaries, so two byte-identical payloads uploaded with different part sizes will look divergent. If your pipeline does multipart uploads, store a content hash in object metadata at write time and compare that instead.
#!/usr/bin/env bash
# Scans an object store for duplicate, DIVERGENT writes to the same logical
# output window: the signature of two task replicas racing during a deploy.
# Works against any S3-compatible store (AWS S3, MinIO, Ceph RGW). It only
# reports, so it is safe to run against production.
# NOTE: ETags are content hashes only for single-part uploads; for multipart
# objects, compare a content hash stored in object metadata instead.
set -euo pipefail
BUCKET=”${1:?usage: detect_divergence.sh <bucket> <prefix>}”
PREFIX=”${2:?usage: detect_divergence.sh <bucket> <prefix>}”
# Object versioning is what makes a duplicate write visible at all: without it,
# the second write silently overwrites the first and you lose the evidence.
versions_json=”$(aws s3api list-object-versions \
–bucket “$BUCKET” –prefix “$PREFIX” \
–query ‘Versions[].{Key:Key,ETag:ETag,Time:LastModified}’ \
–output json)”
# A key with one version, or several versions sharing an ETag, is benign. A key
# with MULTIPLE DISTINCT ETags means two runs produced different bytes for the
# same window: a real correctness defect, not a cosmetic duplicate.
echo “$versions_json” | jq -r ‘
group_by(.Key)[]
| {key: .[0].Key, etags: ([.[].ETag] | unique), writes: length}
| select((.etags | length) > 1)
| “DIVERGENT \(.key) writes=\(.writes) payloads=\(.etags | length)”‘
# Exit non-zero if any divergence was found, so a deploy gate can block.
divergent=”$(echo “$versions_json” | jq ‘
[ group_by(.Key)[] | select(([.[].ETag] | unique | length) > 1) ] | length’)”
echo “scanned prefix=$PREFIX divergent_keys=$divergent”
test “$divergent” -eq 0
Run this on a schedule and wire the exit code into a deployment gate. A nonzero result during or just after a rollout is a direct measurement of the bug. The divergence rate climbs sharply with task duration. A job under 30 seconds rarely overlaps a rollout, while a job in the 2-to-3-minute range will overlap nearly every deployment that lands during its run.
Idempotent Output Keying and Atomic Publish
The fix comes in two parts. First, derive the output key from the inputs rather than from wall-clock time or a process identifier. Two replicas working the same scheduled window must compute the same key, so both replicas target a single object instead of two. Second, publish that object atomically, so a reader never sees a partial write and a duplicate publish becomes a no-op rather than a second racing write.
Start with the key. Build it from the fields that define the unit of work: the pipeline name, the closed time window being summarized and a schema version that you bump only when the output format changes. The schema version matters at deploy time: a new binary that emits a new format gets a different key, so it never collides with the old binary’s output.
use sha2::{Digest, Sha256};
// Two task replicas that pick up the same scheduled window build the SAME
// RunSpec. That property is what the whole scheme relies on.
#[derive(Clone)]
struct RunSpec {
pipeline: String,
window_start_epoch: u64, // closed window, deterministic per schedule tick
window_len_secs: u64,
schema_version: u32, // bump only when the OUTPUT FORMAT changes
}
impl RunSpec {
// Content key derived purely from inputs. Identical inputs -> identical key,
// which is what lets two overlapping runs target one object, not two.
fn output_key(&self) -> String {
let mut h = Sha256::new();
h.update(self.pipeline.as_bytes());
h.update(self.window_start_epoch.to_be_bytes());
h.update(self.window_len_secs.to_be_bytes());
h.update(self.schema_version.to_be_bytes());
let digest = h.finalize();
format!(“{}/{}/state-{:x}”, self.pipeline, self.window_start_epoch, digest)
}
}
Since the key is content-derived, identical inputs produce the same key, and a changed format produces a new one. The second part is publishing without a destructive overwrite. The pattern is to write to a unique temporary object, flush it to durable storage, then promote it into the final key with an operation that’s atomic at the storage layer. On a single filesystem that promotion is rename. On an object store it’s a conditional put that fails if the key already exists, or a multipart completion.
use std::fs;
use std::io::Write;
// Atomic publish: write to a unique temp object, fsync, then promote into the
// final key with an operation that is atomic at the storage layer. On one
// filesystem that is rename(2). On an object store it maps to a conditional
// PutObject (If-None-Match) or a multipart completion, NOT a streamed append.
fn atomic_publish(key: &str, payload: &[u8], writer_id: &str) -> std::io::Result<bool> {
let final_path = store_root().join(key);
fs::create_dir_all(final_path.parent().unwrap())?;
// Skip-if-exists: a duplicate run that finds the object already there does
// no work and produces no second write. Handles the common finish-early case.
if final_path.exists() {
return Ok(false);
}
let tmp = store_root().join(format!(“.tmp-{}-{}”, key.replace(‘/’, “_”), writer_id));
let mut f = fs::File::create(&tmp)?;
f.write_all(payload)?;
f.sync_all()?; // durable before it becomes visible
// Two writers can both pass the exists() check; rename is still atomic, so
// the object is whole, and the payloads are byte-identical because the key
// is content-derived. It does not matter which one lands.
fs::rename(&tmp, &final_path)?;
Ok(true)
}
Skip-if-exists handles the common case where one replica finishes well ahead of the other. The harder case is two writers that both pass the existence check before either commits. Atomic promotion saves you there: the object is always whole, and if the key captures every input that affects the output (the discipline covered in the trade-offs section), both candidate payloads are byte-identical, so it doesn’t matter which one lands.
Readers need one more guarantee. They should never have to guess which key is current. Publish each generation under its own immutable key, then advance a single pointer with a compare-and-swap (CAS), so consumers follow the pointer and always read a complete generation. A losing writer detects the conflict and backs off instead of regressing the pointer to an older or duplicated generation.
use std::fs;
// Readers follow a single pointer, so they always observe one COMPLETE
// generation, never a partially written one.
fn publish_generation(key: &str, payload: &[u8]) -> std::io::Result<()> {
let p = store_root().join(key);
fs::create_dir_all(p.parent().unwrap())?;
fs::write(p, payload) // immutable, content-addressed
}
// Optimistic compare-and-swap: only advance the pointer if it still holds the
// value the writer last observed. A losing writer (a duplicate from the deploy)
// detects the conflict and backs off instead of regressing to an older or
// duplicated generation. Maps to a conditional write (If-Match on an ETag) in a
// real object store or a small consistent key-value store.
fn cas_pointer(expected: Option<&str>, next: &str) -> std::io::Result<bool> {
let ptr = store_root().join(“latest”);
let current = fs::read_to_string(&ptr).ok();
let matches = match (current.as_deref(), expected) {
(None, None) => true,
(Some(c), Some(e)) => c == e,
_ => false,
};
if !matches {
return Ok(false); // someone else moved it; do not clobber
}
fs::write(&ptr, next)?;
Ok(true)
}
Trade-offs: Content Keys Versus Locks, and What Teams Pay
Content-derived keys with atomic publish cost more storage and more writes than a single fixed path. Every generation is retained until a lifecycle policy expires it, and versioning multiplies object count during the overlap windows you can now observe. For a pipeline producing one object per minute the added cost is small, a few percent of the storage line in most setups, and it buys an output history you can audit and roll back.
Against distributed locks the comparison is sharper. A lock-based design adds a hard dependency on a coordination service in the write path, which means you inherit its availability and tail latency. The keying approach has no such dependency at write time. Its correctness comes from determinism and atomic promotion, both properties of code and storage you already run. The cost is that every input that affects the output must be folded into the key, or two genuinely different results can collide under one key and you’re back to silent corruption.
The safe way to adopt this is incremental rollout validated by the detection scan. Deploy the keyed publish path to a single region first, then run the divergence scan across a full deployment cycle before widening. A clean scan across one rollout is strong evidence the keying covers every input that matters. The verification below runs two overlapping replicas of the same task and asserts that exactly one object results and its contents match what either replica intended.
// Verification: two replicas of the SAME logical task, as happens when an old
// pod and a new pod both fire during a rolling deploy. Exactly one object must
// result, and its bytes must match what either replica intended.
fn overlapping_runs_converge() {
let spec = RunSpec {
pipeline: “border-state”.into(),
window_start_epoch: 1_726_000_000,
window_len_secs: 60,
schema_version: 3,
};
let key = spec.output_key();
let payload = build_payload(&spec);
let wrote_old = atomic_publish(&key, &payload, “old-replica”).unwrap();
let wrote_new = atomic_publish(&key, &payload, “new-replica”).unwrap();
assert!(wrote_old ^ wrote_new, “exactly one replica writes the object”);
assert_eq!(walk(&store_root()).len(), 1, “overlap converges to one export”);
}
#[test]
fn identical_inputs_yield_identical_keys() {
let a = RunSpec { pipeline: “p”.into(), window_start_epoch: 100,
window_len_secs: 60, schema_version: 1 };
assert_eq!(a.output_key(), a.clone().output_key());
}
One caveat: this pattern assumes the task’s inputs fully determine its output. If a job reads from a mutable source mid-run, a database row that changes while it computes, no key derivation can fix that. Key the inputs you can, snapshot the ones you can’t, and treat snapshot isolation as a precondition.
Teams that skip this work don’t see failures immediately. The pipeline runs clean for weeks, then a deployment lands during a long task and a single corrupted generation flows downstream. By the time anyone traces the bad decision back to its source, the offending object has been overwritten and the logs show two clean completions. The keying and atomic publish pattern turns that entire class of incident into a no-op, and the detection scan turns the residual risk into a number you can watch.
