kiseki-composition 2026.1.0

Composition context for Kiseki: file/object metadata, namespaces, multipart, versioning.
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

//! Bridge between Composition and Log contexts.
//!
//! When attached to a `CompositionStore`, composition mutations
//! (create, update, delete) emit deltas to the log shard. This
//! wires the Composition → Log data path per api-contracts.md.

use kiseki_common::ids::{ChunkId, NodeId, OrgId, ShardId};
use kiseki_common::time::{ClockQuality, DeltaTimestamp, HybridLogicalClock, WallTime};
use kiseki_log::delta::OperationType;
use kiseki_log::traits::{AppendDeltaRequest, LogOps};

/// Emit a delta to the log for a composition mutation.
///
/// Called by `CompositionStore` after a successful create/update/delete.
/// The payload is the composition ID serialized as bytes (opaque to
/// the log per I-L7).
pub(crate) fn emit_delta<L: LogOps + ?Sized>(
    log: &L,
    shard_id: ShardId,
    tenant_id: OrgId,
    operation: OperationType,
    hashed_key: [u8; 32],
    chunk_refs: Vec<ChunkId>,
    payload: Vec<u8>,
) -> bool {
    let timestamp = now_timestamp();
    let req = AppendDeltaRequest {
        shard_id,
        tenant_id,
        operation,
        timestamp,
        hashed_key,
        chunk_refs,
        payload,
        has_inline_data: false,
    };
    match log.append_delta(req) {
        Ok(_seq) => true,
        Err(e) => {
            // Log the error — callers check the return value to decide
            // whether to roll back the local mutation (PIPE-ADV-1).
            tracing::warn!(error = %e, "delta emission failed");
            false
        }
    }
}

/// Monotonic logical counter for HLC tie-breaking (PIPE-ADV-2).
static HLC_LOGICAL: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);

fn now_timestamp() -> DeltaTimestamp {
    let now_ms = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX));
    let logical = HLC_LOGICAL.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    DeltaTimestamp {
        hlc: HybridLogicalClock {
            physical_ms: now_ms,
            logical,
            node_id: NodeId(0),
        },
        wall: WallTime {
            millis_since_epoch: now_ms,
            timezone: "UTC".into(),
        },
        quality: ClockQuality::Ntp,
    }
}