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

ozzy_journal/operation/
schema.rs
