linkrs-sync 0.1.0

Synchronization primitives for Linkrs
Documentation 
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
107
108
109
110

//! Shared types for the sync manager: error, configs, edge views.
#[cfg(feature = "fulltext")]
use crate::coordinator::CoordinatorError;
use linkrs_core::Value;
#[derive(Debug, Clone)]
pub struct IndexCreateRequest {
    pub space_id: u64,
    pub index_name: String,
    pub schema_name: String,
    pub index_type: String,
    pub fields: Vec<(String, Value)>,
    pub properties: Vec<String>,
}
/// Delivery policy shared by one or more sync manager workers.
#[derive(Debug, Clone)]
pub struct OutboxConsumerConfig {
    pub consumer_id: String,
    pub batch_size: usize,
    pub lease_duration_ms: u64,
    pub max_retries: u64,
    /// Max concurrent `claim → apply → ack` workers per target.
    /// `1` = single-threaded delivery (Local); `4` = Qdrant concurrent.
    pub max_concurrency: usize,
}
impl Default for OutboxConsumerConfig {
    fn default() -> Self {
        Self {
            consumer_id: format!("sync-manager-{}", uuid::Uuid::new_v4()),
            batch_size: 128,
            lease_duration_ms: 30_000,
            max_retries: 16,
            max_concurrency: 1,
        }
    }
}
/// Backpressure limits for the transactional outbox staging path.
#[derive(Debug, Clone)]
pub struct OutboxBackpressureConfig {
    /// Maximum intents staged for a single transaction.
    pub max_pending_per_txn: usize,
    /// Maximum intents staged across all in-flight transactions plus the
    /// durable `pending` backlog (queried lazily). Exceeding either limit
    /// makes `stage_intent` return `OutboxBackpressure`.
    pub max_pending_total: usize,
}
impl Default for OutboxBackpressureConfig {
    fn default() -> Self {
        Self {
            max_pending_per_txn: 10_000,
            max_pending_total: 100_000,
        }
    }
}
#[derive(Debug, Clone)]
pub struct EdgeRef<'a> {
    pub src: &'a Value,
    pub dst: &'a Value,
    pub edge_type: &'a str,
}
impl<'a> EdgeRef<'a> {
    pub fn new(src: &'a Value, dst: &'a Value, edge_type: &'a str) -> Self {
        Self {
            src,
            dst,
            edge_type,
        }
    }

    pub fn id(&self) -> String {
        format!("{}->{}", self.src, self.dst)
    }
}
#[derive(Debug, Clone)]
pub struct EdgeProps<'a> {
    pub old: &'a [(String, Value)],
    pub new: &'a [(String, Value)],
}
impl<'a> EdgeProps<'a> {
    pub fn new(old: &'a [(String, Value)], new: &'a [(String, Value)]) -> Self {
        Self { old, new }
    }
}
#[derive(Debug, thiserror::Error)]
pub enum SyncError {
    #[cfg(feature = "fulltext")]
    #[error("Coordinator error: {0}")]
    CoordinatorError(#[from] CoordinatorError),

    #[cfg(feature = "fulltext")]
    #[error("Sync coordinator error: {0}")]
    SyncCoordinatorError(#[from] crate::coordinator::SyncCoordinatorError),

    #[error("Buffer error: {0}")]
    BufferError(String),

    #[error("Vector error: {0}")]
    VectorError(String),

    #[error("Persistence error: {0}")]
    PersistenceError(String),

    #[error("Outbox backpressure: {0}")]
    OutboxBackpressure(String),

    #[error("Rebuild busy: {0}")]
    RebuildBusy(String),

    #[error("Internal error: {0}")]
    Internal(String),
}