ozzy_journal/operation/
schema.rs

1//! Canonical identities, operation bodies, and admission metadata.
2
3mod append;
4pub use append::{
5    Append, AppendBatch, AppendBatchSummary, AppendRecord, AppendSummary, RecordPosition,
6    ResolvedAppend,
7};
8
9use super::OperationCodecError;
10use ozzy_proto::{
11    ConsumerGroupId, ConsumerMemberId, GroupId, Offset, OperationId, OwnerEpoch, PartitionId,
12    PartitionIncarnation, ProducerEpoch, ProducerId, ProducerSequence, SubscriptionId,
13};
14use std::num::NonZeroU64;
15
16/// A fixed-width digest or externally supplied identity fingerprint.
17/// Computed integrity digests contain XXH3-128 followed by 16 zero bytes;
18/// see [`crate::integrity`]. External principal fingerprints remain opaque.
19#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
20pub struct Digest([u8; 32]);
21
22impl Digest {
23    /// Digest used by the configured genesis operation prefix.
24    pub const ZERO: Self = Self([0; 32]);
25
26    /// Construct a digest from its canonical bytes.
27    pub const fn from_bytes(bytes: [u8; 32]) -> Self {
28        Self(bytes)
29    }
30
31    /// Return canonical digest bytes.
32    pub const fn as_bytes(&self) -> &[u8; 32] {
33        &self.0
34    }
35}
36
37/// One operation kind in the canonical journal schema.
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39#[repr(u16)]
40pub enum OperationKind {
41    /// Create an incarnation and bind its logical address and owner fence.
42    CreatePartition = 1,
43    /// Open or fence one partition-local producer session.
44    OpenProducer = 2,
45    /// Assign consecutive records to producer sequences and partition offsets.
46    Append = 3,
47    /// Declare exclusive individual or consumer-group progress.
48    Progress = 4,
49    /// Install or revoke a fenced consumer-group partition assignment.
50    Assign = 5,
51    /// Advance the partition retention floor.
52    Trim = 6,
53    /// Install a newer retention-policy revision.
54    PartitionPolicy = 7,
55    /// Record a control-operation identity without changing record state.
56    Barrier = 8,
57    /// Advance the retained producer retry-result floor.
58    ProducerResultFloor = 9,
59    /// Resolve producer attempts independently of newly stored records.
60    ResolvedAppend = 10,
61    /// Advance replicated logical time while no unique records are arriving.
62    DedupWatermark = 11,
63    /// Persist coordinator membership and a source ownership plan.
64    GroupPlan = 12,
65    /// Retire one coordinator group incarnation and free its name.
66    GroupDelete = 13,
67    /// Remove one idle producer session and its retry results.
68    RetireProducer = 14,
69}
70
71impl TryFrom<u16> for OperationKind {
72    type Error = OperationCodecError;
73
74    fn try_from(value: u16) -> Result<Self, Self::Error> {
75        match value {
76            1 => Ok(Self::CreatePartition),
77            2 => Ok(Self::OpenProducer),
78            3 => Ok(Self::Append),
79            4 => Ok(Self::Progress),
80            5 => Ok(Self::Assign),
81            6 => Ok(Self::Trim),
82            7 => Ok(Self::PartitionPolicy),
83            8 => Ok(Self::Barrier),
84            9 => Ok(Self::ProducerResultFloor),
85            10 => Ok(Self::ResolvedAppend),
86            11 => Ok(Self::DedupWatermark),
87            12 => Ok(Self::GroupPlan),
88            13 => Ok(Self::GroupDelete),
89            14 => Ok(Self::RetireProducer),
90            other => Err(OperationCodecError::UnsupportedOperationKind(other)),
91        }
92    }
93}
94
95impl OperationKind {
96    /// Whether every valid body of this kind returns an explicit control
97    /// identity from `OperationBody::operation_id`. Admission uses it to skip
98    /// decoding APPEND-like bodies that canonical validation decodes anyway,
99    /// and reserves identity capacity only for these kinds.
100    pub const fn has_operation_id(self) -> bool {
101        !matches!(
102            self,
103            Self::CreatePartition
104                | Self::Append
105                | Self::Progress
106                | Self::ResolvedAppend
107                | Self::DedupWatermark
108                | Self::RetireProducer
109        )
110    }
111}
112
113/// Expected logical chain position before an operation.
114#[derive(Debug, Clone, Copy, PartialEq, Eq)]
115pub struct ChainPosition {
116    next_op_number: u64,
117    previous_digest: Digest,
118}
119
120impl ChainPosition {
121    /// Configured genesis position. Journal operations begin at one.
122    pub const GENESIS: Self = Self {
123        next_op_number: 1,
124        previous_digest: Digest::ZERO,
125    };
126
127    /// Bind the next canonical operation number to its predecessor digest.
128    pub const fn new(next_op_number: u64, previous_digest: Digest) -> Self {
129        Self {
130            next_op_number,
131            previous_digest,
132        }
133    }
134
135    /// Next expected canonical operation number.
136    pub const fn next_op_number(self) -> u64 {
137        self.next_op_number
138    }
139
140    /// Canonical chain digest immediately before the next operation.
141    pub const fn previous_digest(self) -> Digest {
142        self.previous_digest
143    }
144}
145
146/// One canonical operation ready for wire or physical-disk framing.
147#[derive(Debug, Clone, Copy, PartialEq, Eq)]
148pub struct CanonicalOperation<'a> {
149    /// Persistent partition replication-group identity.
150    pub group_id: GroupId,
151    /// Membership/configuration fence at admission.
152    pub configuration_epoch: u64,
153    /// Original admission view, preserved through replication and repair.
154    pub original_view: u64,
155    /// Partition-local canonical operation number.
156    pub op_number: u64,
157    /// Canonical chain digest immediately before this operation.
158    pub previous_digest: Digest,
159    /// Canonical operation discriminator.
160    pub kind: OperationKind,
161    /// Exact canonical body bytes covered by logical integrity.
162    pub body: &'a [u8],
163}
164
165/// Logical identity without any assumption about body storage or contiguity.
166#[derive(Debug, Clone, Copy, PartialEq, Eq)]
167pub struct OperationHeader {
168    /// Persistent partition replication-group identity.
169    pub group_id: GroupId,
170    /// Membership/configuration fence at admission.
171    pub configuration_epoch: u64,
172    /// Original admission view, preserved through replication and repair.
173    pub original_view: u64,
174    /// Partition-local canonical operation number.
175    pub op_number: u64,
176    /// Canonical chain digest immediately before this operation.
177    pub previous_digest: Digest,
178    /// Canonical operation discriminator.
179    pub kind: OperationKind,
180}
181
182impl CanonicalOperation<'_> {
183    /// Borrow-independent canonical metadata for this exact operation.
184    pub const fn header(&self) -> OperationHeader {
185        OperationHeader {
186            group_id: self.group_id,
187            configuration_epoch: self.configuration_epoch,
188            original_view: self.original_view,
189            op_number: self.op_number,
190            previous_digest: self.previous_digest,
191            kind: self.kind,
192        }
193    }
194}
195
196/// Independent age/size retention targets. Neither target means keep forever.
197#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
198pub struct RetentionPolicy {
199    /// Optional maximum retained record age in milliseconds.
200    pub max_age_millis: Option<NonZeroU64>,
201    /// Optional target for selected segment capacities, including active space.
202    pub max_bytes: Option<NonZeroU64>,
203}
204
205/// Partition creation. Policy revision starts at one implicitly.
206#[derive(Debug, Clone, PartialEq, Eq)]
207pub struct CreatePartition<'a> {
208    /// Exact partition record incarnation.
209    pub partition: PartitionIncarnation,
210    /// Logical stream namespace.
211    pub stream: &'a str,
212    /// Logical topic name.
213    pub topic: &'a str,
214    /// Partition number within its topic.
215    pub partition_id: PartitionId,
216    /// Expected partition ownership fence.
217    pub owner_epoch: OwnerEpoch,
218    /// Immutable first-wins deduplication window in replicated log milliseconds.
219    /// Absent disables event deduplication for this partition incarnation.
220    pub dedup_window_millis: Option<NonZeroU64>,
221    /// Retention policy stored in canonical state.
222    pub retention: RetentionPolicy,
223}
224
225/// Idempotent producer-session open or fencing transition.
226#[derive(Debug, Clone, Copy, PartialEq, Eq)]
227pub struct OpenProducer {
228    /// Exact partition record incarnation.
229    pub partition: PartitionIncarnation,
230    /// Producer identity scoped to this partition.
231    pub producer_id: ProducerId,
232    /// Expected existing session epoch; none requires no session.
233    pub expected_epoch: Option<ProducerEpoch>,
234    /// New producer fence, strictly newer than an existing session.
235    pub new_epoch: ProducerEpoch,
236    /// Exact control-operation retry identity.
237    pub operation_id: OperationId,
238}
239
240/// Stable owner of one processing-progress cursor.
241#[derive(Debug, Clone, Copy, PartialEq, Eq)]
242pub enum ProgressOwner {
243    /// Progress declared by one standalone subscription.
244    Subscription(SubscriptionId),
245    /// Progress declared by a fenced consumer-group assignment.
246    ConsumerGroup(ConsumerGroupId),
247}
248
249/// Source-partition ownership required to declare consumer-group progress.
250#[derive(Debug, Clone, Copy, PartialEq, Eq)]
251pub struct ProgressAssignment {
252    /// Exact nonzero source-partition assignment epoch.
253    pub epoch: u64,
254    /// Member whose confirmed assignment permits processing progress.
255    pub member: ConsumerMemberId,
256}
257
258/// Durable contiguous consumer-processing progress transition.
259#[derive(Debug, Clone, Copy, PartialEq, Eq)]
260pub struct Progress {
261    /// Standalone subscription or fenced consumer group declaring progress.
262    pub owner: ProgressOwner,
263    /// Exact partition record incarnation.
264    pub partition: PartitionIncarnation,
265    /// Member and assignment fence required for consumer-group progress.
266    pub assignment: Option<ProgressAssignment>,
267    /// Expected prior exclusive offset; none requires no progress entry.
268    pub expected_progress: Option<Offset>,
269    /// New exclusive progress offset within confirmed record state.
270    pub new_progress: Offset,
271    /// Exact control-operation retry identity.
272    pub operation_id: OperationId,
273}
274
275/// Consumer-group partition assignment transition.
276#[derive(Debug, Clone, Copy, PartialEq, Eq)]
277pub struct Assign {
278    /// Consumer group whose assignment changes.
279    pub consumer_group_id: ConsumerGroupId,
280    /// Exact partition record incarnation.
281    pub partition: PartitionIncarnation,
282    /// Expected current assignment fence.
283    pub expected_assignment_epoch: u64,
284    /// New assignment fence, strictly newer than the current one.
285    pub new_assignment_epoch: u64,
286    /// Committed coordinator plan that authorizes this transition.
287    pub plan_revision: u64,
288    /// Initial group cursor, or the exact existing progress captured at handoff.
289    pub resume_cursor: Offset,
290    /// New assigned member; none revokes ownership.
291    pub new_member: Option<ConsumerMemberId>,
292    /// Exact control-operation retry identity.
293    pub operation_id: OperationId,
294}
295
296/// Advance one partition's committed earliest-retained offset.
297#[derive(Debug, Clone, Copy, PartialEq, Eq)]
298pub struct Trim {
299    /// Exact partition record incarnation.
300    pub partition: PartitionIncarnation,
301    /// Expected current retention floor.
302    pub expected_floor: Offset,
303    /// New monotonic retention floor.
304    pub new_floor: Offset,
305    /// Exact control-operation retry identity.
306    pub operation_id: OperationId,
307}
308
309/// Replace one partition's retention policy by revision.
310#[derive(Debug, Clone, Copy, PartialEq, Eq)]
311pub struct PartitionPolicy {
312    /// Exact partition record incarnation.
313    pub partition: PartitionIncarnation,
314    /// Expected current retention-policy revision.
315    pub expected_revision: u64,
316    /// New retention-policy revision.
317    pub new_revision: u64,
318    /// Retention policy stored in canonical state.
319    pub retention: RetentionPolicy,
320    /// Exact control-operation retry identity.
321    pub operation_id: OperationId,
322}
323
324/// Ordered no-op used to establish an authority or read boundary.
325#[derive(Debug, Clone, Copy, PartialEq, Eq)]
326pub struct Barrier {
327    /// Exact control-operation retry identity.
328    pub operation_id: OperationId,
329}
330
331/// Advance the exclusive floor below which producer retry results expired.
332#[derive(Debug, Clone, Copy, PartialEq, Eq)]
333pub struct ProducerResultFloor {
334    /// Exact partition record incarnation.
335    pub partition: PartitionIncarnation,
336    /// Producer identity scoped to this partition.
337    pub producer_id: ProducerId,
338    /// Expected producer-session fence.
339    pub producer_epoch: ProducerEpoch,
340    /// Expected retained producer retry-result floor.
341    pub expected_floor: ProducerSequence,
342    /// New monotonic producer retry-result floor.
343    pub new_floor: ProducerSequence,
344    /// Exact control-operation retry identity.
345    pub operation_id: OperationId,
346}
347
348/// Primary-resolved monotonic deduplication time, independent of wall clocks.
349#[derive(Debug, Clone, Copy, PartialEq, Eq)]
350pub struct DedupWatermark {
351    /// Exact partition record incarnation.
352    pub partition: PartitionIncarnation,
353    /// Expected partition ownership fence.
354    pub owner_epoch: OwnerEpoch,
355    /// Replicated logical time in milliseconds, strictly newer than current time.
356    pub logical_time: u64,
357}
358
359/// Primary-chosen removal of one idle producer session. The expected epoch
360/// and next sequence fence any open or append ordered before it.
361#[derive(Debug, Clone, Copy, PartialEq, Eq)]
362pub struct RetireProducer {
363    /// Exact partition record incarnation.
364    pub partition: PartitionIncarnation,
365    /// Producer identity scoped to this partition.
366    pub producer_id: ProducerId,
367    /// Expected producer-session fence.
368    pub producer_epoch: ProducerEpoch,
369    /// Expected next producer sequence.
370    pub next_sequence: ProducerSequence,
371}
372
373/// One typed canonical operation body.
374#[derive(Debug, Clone, PartialEq, Eq)]
375pub enum OperationBody<'a> {
376    /// Create an incarnation and bind its logical address and owner fence.
377    CreatePartition(CreatePartition<'a>),
378    /// Open or fence one partition-local producer session.
379    OpenProducer(OpenProducer),
380    /// Assign consecutive records to producer sequences and partition offsets.
381    Append(Append<'a>),
382    /// Resolve producer attempts independently of newly stored records.
383    ResolvedAppend(ResolvedAppend<'a>),
384    /// Declare exclusive individual or consumer-group progress.
385    Progress(Progress),
386    /// Install or revoke a fenced consumer-group partition assignment.
387    Assign(Assign),
388    /// Advance the partition retention floor.
389    Trim(Trim),
390    /// Install a newer retention-policy revision.
391    PartitionPolicy(PartitionPolicy),
392    /// Record a control-operation identity without changing record state.
393    Barrier(Barrier),
394    /// Advance the retained producer retry-result floor.
395    ProducerResultFloor(ProducerResultFloor),
396    /// Advance replicated logical time without changing record coordinates.
397    DedupWatermark(DedupWatermark),
398    /// Persist coordinator membership before changing source ownership.
399    GroupPlan(super::GroupPlan<'a>),
400    /// Retire one coordinator group incarnation and free its name.
401    GroupDelete(super::GroupDelete),
402    /// Remove one idle producer session and its retry results.
403    RetireProducer(RetireProducer),
404}
405
406impl OperationBody<'_> {
407    /// Explicit retry identity for control operations. APPEND identity belongs
408    /// to its producer sequences; partition creation uses its incarnation.
409    /// Progress retries resolve by their cumulative cursor, so its request ID
410    /// is not an identity.
411    pub const fn operation_id(&self) -> Option<OperationId> {
412        match self {
413            Self::CreatePartition(_)
414            | Self::Append(_)
415            | Self::Progress(_)
416            | Self::ResolvedAppend(_)
417            | Self::DedupWatermark(_)
418            | Self::RetireProducer(_) => None,
419            Self::OpenProducer(value) => Some(value.operation_id),
420            Self::Assign(value) => Some(value.operation_id),
421            Self::Trim(value) => Some(value.operation_id),
422            Self::PartitionPolicy(value) => Some(value.operation_id),
423            Self::Barrier(value) => Some(value.operation_id),
424            Self::ProducerResultFloor(value) => Some(value.operation_id),
425            Self::GroupPlan(value) => Some(value.operation_id),
426            Self::GroupDelete(value) => Some(value.operation_id),
427        }
428    }
429
430    /// Return the canonical operation discriminator.
431    pub const fn kind(&self) -> OperationKind {
432        match self {
433            Self::CreatePartition(_) => OperationKind::CreatePartition,
434            Self::OpenProducer(_) => OperationKind::OpenProducer,
435            Self::Append(_) => OperationKind::Append,
436            Self::ResolvedAppend(_) => OperationKind::ResolvedAppend,
437            Self::Progress(_) => OperationKind::Progress,
438            Self::Assign(_) => OperationKind::Assign,
439            Self::Trim(_) => OperationKind::Trim,
440            Self::PartitionPolicy(_) => OperationKind::PartitionPolicy,
441            Self::Barrier(_) => OperationKind::Barrier,
442            Self::ProducerResultFloor(_) => OperationKind::ProducerResultFloor,
443            Self::DedupWatermark(_) => OperationKind::DedupWatermark,
444            Self::GroupPlan(_) => OperationKind::GroupPlan,
445            Self::GroupDelete(_) => OperationKind::GroupDelete,
446            Self::RetireProducer(_) => OperationKind::RetireProducer,
447        }
448    }
449}
450
451/// Resource bounds applied to canonical operation bodies.
452#[derive(Debug, Clone, Copy, PartialEq, Eq)]
453pub struct OperationLimits {
454    /// Maximum encoded canonical operation-body bytes.
455    pub max_body_bytes: usize,
456    /// Maximum UTF-8 bytes per stream or topic name.
457    pub max_name_bytes: usize,
458    /// Maximum independently addressed batches per APPEND.
459    pub max_append_batches: usize,
460    /// Maximum combined records per APPEND.
461    pub max_records: usize,
462    /// Maximum combined payload parts per APPEND.
463    pub max_parts: usize,
464    /// Maximum combined APPEND payload bytes.
465    pub max_payload_bytes: usize,
466    /// Maximum members in one coordinator intent.
467    pub max_group_members: usize,
468    /// Maximum source partitions in one coordinator intent.
469    pub max_group_partitions: usize,
470}
471
472impl Default for OperationLimits {
473    fn default() -> Self {
474        Self {
475            max_body_bytes: 64 * 1024 * 1024,
476            max_name_bytes: 255,
477            max_append_batches: 1_024,
478            max_records: 1_024,
479            max_parts: 65_536,
480            max_payload_bytes: 64 * 1024 * 1024,
481            max_group_members: 256,
482            max_group_partitions: 1024,
483        }
484    }
485}
ozzy_journal/operation/schema.rs

ozzy_journal/operation/
schema.rs
