ursula_stream/

command.rs

use std::fmt;

use bytes::Bytes;
use serde::Deserialize;
use serde::Serialize;
use ursula_shard::BucketStreamId;

use crate::model::ColdChunkRef;
use crate::model::ExternalPayloadRef;
use crate::model::ProducerRequest;
use crate::model::StreamAttrs;
use crate::snapshot::StreamSnapshot;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum StreamCommand {
    CreateBucket {
        bucket_id: String,
    },
    DeleteBucket {
        bucket_id: String,
    },
    CreateStream {
        stream_id: BucketStreamId,
        content_type: String,
        initial_payload: Bytes,
        close_after: bool,
        stream_seq: Option<String>,
        producer: Option<ProducerRequest>,
        stream_ttl_seconds: Option<u64>,
        stream_expires_at_ms: Option<u64>,
        // `default` keeps pre-attrs replicated records decodable.
        #[serde(default)]
        attrs: Option<StreamAttrs>,
        now_ms: u64,
    },
    CreateExternal {
        stream_id: BucketStreamId,
        content_type: String,
        initial_payload: ExternalPayloadRef,
        #[serde(default)]
        record_ends: Vec<u64>,
        close_after: bool,
        stream_seq: Option<String>,
        producer: Option<ProducerRequest>,
        stream_ttl_seconds: Option<u64>,
        stream_expires_at_ms: Option<u64>,
        // `default` keeps pre-attrs replicated records decodable.
        #[serde(default)]
        attrs: Option<StreamAttrs>,
        now_ms: u64,
    },
    Append {
        stream_id: BucketStreamId,
        content_type: Option<String>,
        payload: Bytes,
        close_after: bool,
        stream_seq: Option<String>,
        producer: Option<ProducerRequest>,
        now_ms: u64,
        record_match: Option<u64>,
    },
    AppendExternal {
        stream_id: BucketStreamId,
        content_type: Option<String>,
        payload: ExternalPayloadRef,
        #[serde(default)]
        record_ends: Vec<u64>,
        close_after: bool,
        stream_seq: Option<String>,
        producer: Option<ProducerRequest>,
        now_ms: u64,
        record_match: Option<u64>,
    },
    AppendBatch {
        stream_id: BucketStreamId,
        content_type: Option<String>,
        payloads: Vec<Bytes>,
        producer: Option<ProducerRequest>,
        now_ms: u64,
    },
    PublishSnapshot {
        stream_id: BucketStreamId,
        snapshot_offset: u64,
        content_type: String,
        payload: Bytes,
        #[serde(default)]
        expected_digest: Option<String>,
        now_ms: u64,
    },
    AdvanceRetention {
        stream_id: BucketStreamId,
        retained_offset: u64,
        now_ms: u64,
    },
    TouchStreamAccess {
        stream_id: BucketStreamId,
        now_ms: u64,
        renew_ttl: bool,
    },
    UpdateStreamAttrs {
        stream_id: BucketStreamId,
        attrs: Option<StreamAttrs>,
        now_ms: u64,
    },
    FlushCold {
        stream_id: BucketStreamId,
        chunk: ColdChunkRef,
    },
    /// Replaces a contiguous run of immutable cold chunks with one equivalent
    /// object. The external cold-index page update is completed before this
    /// command is replicated; applying it schedules the old paths for delayed
    /// reclamation.
    CompactCold {
        stream_id: BucketStreamId,
        old_chunks: Vec<ColdChunkRef>,
        replacement: ColdChunkRef,
        gc_not_before_ms: u64,
    },
    Close {
        stream_id: BucketStreamId,
        stream_seq: Option<String>,
        producer: Option<ProducerRequest>,
        now_ms: u64,
    },
    DeleteStream {
        stream_id: BucketStreamId,
    },
    /// Administrator-triggered tenant offboarding: removes every stream in
    /// the bucket, the bucket itself, and its quota in this group. Monotonic
    /// aggregate usage is retained for asynchronous accounting. Idempotent —
    /// purging an absent bucket reports zero removals.
    PurgeBucket {
        bucket_id: String,
    },
    /// Confirms the leader's background worker has physically reclaimed every
    /// queued cold-GC entry with `seq <= up_to_seq`; removes them from the
    /// replicated queue. Idempotent under replay.
    AckColdGc {
        up_to_seq: u64,
    },
    /// Replaces this group's entire state with a backup snapshot.
    ///
    /// Restore-only: the target group must be empty. Travelling as a normal
    /// replicated command keeps every replica of the restored cluster
    /// deterministic while the cluster retains its own raft identity and
    /// membership -- nothing from the backed-up cluster's raft metadata is
    /// reused.
    ImportSnapshot {
        snapshot: Box<StreamSnapshot>,
    },
    /// Sets or clears this group's data-plane quota record for a bucket.
    /// Replicated to every group so each enforces the same local backstop;
    /// both limits `None` removes the record. Idempotent under replay.
    SetBucketQuota {
        bucket_id: String,
        max_streams: Option<u64>,
        max_retained_bytes: Option<u64>,
    },
}

impl fmt::Display for StreamCommand {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::CreateBucket { bucket_id } => write!(f, "create_bucket:{bucket_id}"),
            Self::DeleteBucket { bucket_id } => write!(f, "delete_bucket:{bucket_id}"),
            Self::CreateStream { stream_id, .. } => write!(f, "create_stream:{stream_id}"),
            Self::CreateExternal {
                stream_id,
                initial_payload,
                ..
            } => write!(
                f,
                "create_external:{stream_id}:{} bytes",
                initial_payload.payload_len
            ),
            Self::Append {
                stream_id, payload, ..
            } => write!(f, "append:{stream_id}:{} bytes", payload.len()),
            Self::AppendExternal {
                stream_id, payload, ..
            } => write!(
                f,
                "append_external:{stream_id}:{} bytes",
                payload.payload_len
            ),
            Self::AppendBatch {
                stream_id,
                payloads,
                ..
            } => write!(f, "append_batch:{stream_id}:{} items", payloads.len()),
            Self::PublishSnapshot {
                stream_id,
                snapshot_offset,
                payload,
                ..
            } => write!(
                f,
                "publish_snapshot:{stream_id}:{snapshot_offset}:{} bytes",
                payload.len()
            ),
            Self::AdvanceRetention {
                stream_id,
                retained_offset,
                ..
            } => write!(f, "advance_retention:{stream_id}:{retained_offset}"),
            Self::TouchStreamAccess {
                stream_id,
                renew_ttl,
                ..
            } => write!(f, "touch_stream_access:{stream_id}:renew_ttl={renew_ttl}"),
            Self::UpdateStreamAttrs { stream_id, .. } => {
                write!(f, "update_stream_attrs:{stream_id}")
            }
            Self::FlushCold { stream_id, chunk } => write!(
                f,
                "flush_cold:{stream_id}:{}..{}",
                chunk.start_offset, chunk.end_offset
            ),
            Self::CompactCold {
                stream_id,
                old_chunks,
                replacement,
                ..
            } => write!(
                f,
                "compact_cold:{stream_id}:{} chunks:{}..{}",
                old_chunks.len(),
                replacement.start_offset,
                replacement.end_offset
            ),
            Self::Close { stream_id, .. } => write!(f, "close_stream:{stream_id}"),
            Self::DeleteStream { stream_id } => write!(f, "delete_stream:{stream_id}"),
            Self::PurgeBucket { bucket_id } => write!(f, "purge_bucket:{bucket_id}"),
            Self::AckColdGc { up_to_seq } => write!(f, "ack_cold_gc:up_to_seq={up_to_seq}"),
            Self::ImportSnapshot { snapshot } => write!(
                f,
                "import_snapshot:buckets={}:streams={}",
                snapshot.buckets.len(),
                snapshot.streams.len()
            ),
            Self::SetBucketQuota { bucket_id, .. } => {
                write!(f, "set_bucket_quota:{bucket_id}")
            }
        }
    }
}