1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
//! Coordinated object-storage (S3) backfill source for Spate.
//!
//! Point the source at a bucket/prefix and it streams every object's
//! records through the pipeline: a **bounded** job that self-terminates
//! (the pipeline exits [`Completed`](spate_core::pipeline::ExitState)) once
//! the prefix is exhausted. Delivery is at-least-once: a resume replays
//! from the last committed position, so duplicates are possible and loss
//! is not.
//!
//! # Shape
//!
//! - **Work is planned as splits.** The fleet's elected leader lists the
//! prefix once and packs the sorted listing into **splits** of ~64 MiB by
//! default: small batches of whole objects, or one byte range of a large
//! uncompressed object. Each has a deterministic identity
//! ([`split_id_for`], [`split_id_for_range`]) and a self-contained
//! [`SplitDescriptor`] carrying its member keys, sizes, and ETags, and its
//! range. Workers lease splits through the coordination store and read
//! them straight from the descriptors: **workers never list**.
//! - **One lane per in-flight split.** A gained split materializes one
//! data lane (one framework partition, one monotonic offset stream); a
//! record's `i64` offset packs (member ordinal within the split, record
//! index within the object, or within the range for a split over a byte
//! range). `coordination.max_in_flight` bounds the working set and
//! therefore read parallelism.
//! - **Progress lives in the coordination store, nowhere else.** Commits
//! are fenced per-split writes; a lost or stolen split resumes on its
//! next owner from the acked watermark, drift-checked against the
//! descriptor's ETag pins. Point the pipeline's `coordination:` section
//! at a shared store (or hand in a coordinator built in code via
//! [`S3Source::with_coordinator`]) and every replica of the pipeline
//! shares the backfill, takes a leader-assigned share, and takes over
//! from the dead. Without one the source runs solo over an in-process
//! store with default tuning, and `store: { memory: {} }` runs that store
//! with the section's tuning: correct, but a restart replays the prefix (a
//! startup WARN says so). This crate names no concrete backend and has no store feature.
//! - **Bad objects poison their split, not the pipeline.** An object
//! deleted after planning, overwritten under its ETag pin, corrupt, or
//! unreadable past the retry budget hands its split back (surfaced in
//! `spate_s3_source_objects_failed_total{reason}`); at the attempt cap
//! the split is quarantined. A bounded job with quarantined splits ends
//! **failed**, never silently incomplete. A rejected object read and a
//! client-configuration error are immediately fatal.
//! - **Records are framed, not decoded, here.** The source is
//! format-agnostic: it streams object bytes (after gzip/zstd decompression)
//! through a [`RecordFramer`](spate_core::framing::RecordFramer) *you supply*
//! for the objects' format via
//! [`S3Source::with_framer`](crate::S3Source::with_framer) (e.g.
//! `spate-json`'s `NdjsonFramer` for NDJSON), emitting one raw payload per
//! framed record. Deserialization then stays in the operator chain
//! (`spate-json` etc.), as with the Kafka source.
//!
//! # Split identity and drift
//!
//! A split's id digests its sorted member keys **and ETags** (plus the
//! packing-algorithm version, and a byte-range split's range), and every GET
//! is pinned `If-Match` to the descriptor's ETag. The consequences:
//!
//! - Replanning an unchanged prefix reproduces identical ids, so replans
//! are create-if-absent no-ops and completed splits stay completed.
//! - An **overwritten** object shows up as a new split (new id) on the
//! next plan; under an in-flight split it trips the `If-Match` pin and
//! poisons the split. Either way it can never silently splice into
//! committed progress.
//! - A **deleted** object poisons only the splits that still needed it.
//!
//! Prefer an append-only prefix for the lifetime of a backfill; by
//! default the listing is taken once (`refresh_listing: false`), so
//! late-arriving keys are not part of the job.
//!
//! No `object_store` types appear in this crate's public API (the same
//! dependency policy that keeps rdkafka out of `spate-kafka`'s). The one
//! seam that would break it, `S3Source::with_store`, is gated behind the
//! off-by-default `testing` feature so tests can inject wrapped or
//! fault-injecting stores without that type reaching a real consumer.
pub use ;
pub use ;
pub use S3Source;
pub use ;