coven_database/gate/
audience.rs

1//! Audience resolution and atomic partitioning of one captured host changeset.
2
3use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
4
5use rusqlite::ffi;
6use rusqlite::Connection;
7
8use super::ffi::{collect_deletes, for_each_change, ChangeRow, Changegroup};
9use super::model::{
10    fk_column_ref, foreign_keys, rows_referencing, truthy, Gates, SharedRows, TableGate,
11};
12use super::outbound::{
13    deleted_or_live_parent, fk_parent_row, full_state_diff, gate_store_outbound,
14    pre_write_full_state_diff, query_column_present, query_column_text, row_id_for_column_value,
15    DeletedAudiences, DeletedParent, FkParentRow, FullStateDirection, UnresolvedAudience,
16};
17use super::{
18    all_row_ids, query_mapped_rows, query_row_optional, CircleControlFailure, GateError,
19    UnsharedForeignKeyParent,
20};
21use crate::quote_ident;
22use coven_protocol::circle::{
23    row_routing_id, Audience, CircleControlCoord, CircleId, RowRoutingKey,
24};
25use coven_protocol::circle_activation::CircleCurrentState;
26
27mod inbound;
28mod partitioning;
29mod routing;
30mod snapshot_pruning;
31
32pub use inbound::store_audience_transitions;
33pub(crate) use inbound::{
34    align_inbound_scoped_root_audiences, filter_inbound_circle_changeset,
35    filter_inbound_store_rows, filter_snapshot_circle_changeset, normalize_inbound_store_changeset,
36};
37pub(crate) use partitioning::{
38    audience_moves, partition_outbound, validate_accepted_foreign_key_closure,
39    validate_scoped_foreign_key_audiences,
40};
41pub(crate) use routing::{active_circle_control, capture_routing_changes, live_row_audience};
42pub(crate) use snapshot_pruning::{
43    prune_ineligible_scoped_rows, prune_private_routes_without_rows, retain_projection_rows,
44    retain_snapshot_audience_rows, validate_snapshot_routing_state,
45};
46
47pub fn is_routing_table(table: &str) -> bool {
48    matches!(table, "_coven_audience" | "_coven_row_routes")
49}
50
51/// Remove capture-generated routing rows before reapplying the host's recorded
52/// edits. Routing is derived again from those edits on the accepted projection.
53pub(crate) fn recorded_host_changeset(changeset: &[u8]) -> Result<Vec<u8>, GateError> {
54    let group = Changegroup::new()?;
55    unsafe {
56        for_each_change(changeset, |iter, row| {
57            if !is_routing_table(&row.table) {
58                group.add_change(iter)?;
59            }
60            Ok(())
61        })?;
62    }
63    group.output()
64}
65
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub struct AudiencePartition {
68    pub audience: Audience,
69    pub control: Option<CirclePartitionControl>,
70    pub changeset: Vec<u8>,
71}
72
73#[derive(Debug, Clone, PartialEq, Eq)]
74pub struct AudienceMove {
75    pub source: Audience,
76    pub destination: Audience,
77    pub rows: BTreeSet<(String, String)>,
78    /// The moved row's `_updated_at` after the change that moved it — the version
79    /// at which its whole component now lives in `destination`, and the stamp the
80    /// routing transitions for that component carry.
81    pub stamp: String,
82}
83
84#[derive(Debug, Clone, PartialEq, Eq)]
85pub(crate) struct PartitionedAudienceWrite {
86    pub partitions: Vec<AudiencePartition>,
87    pub moves: Vec<AudienceMove>,
88}
89
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub struct CirclePartitionControl {
92    coordinate: CircleControlCoord,
93    stored_json: String,
94}
95
96#[derive(Debug, thiserror::Error)]
97pub enum CirclePartitionControlError {
98    #[error("parse Circle partition control: {0}")]
99    Json(#[from] serde_json::Error),
100    #[error("invalid Circle partition control: {0}")]
101    Control(#[from] coven_protocol::circle_control::CircleControlCoordError),
102}
103
104impl CirclePartitionControl {
105    pub fn from_stored_json(stored_json: String) -> Result<Self, CirclePartitionControlError> {
106        let coordinate: CircleControlCoord = serde_json::from_str(&stored_json)?;
107        coordinate.validate()?;
108        Ok(Self {
109            coordinate,
110            stored_json,
111        })
112    }
113
114    pub fn coordinate(&self) -> &CircleControlCoord {
115        &self.coordinate
116    }
117
118    pub fn stored_json(&self) -> &str {
119        &self.stored_json
120    }
121}
122
123pub struct RoutingChanges {
124    store_mirror: Vec<u8>,
125    private_routes: BTreeMap<Audience, Vec<u8>>,
126    deleted_rows: BTreeMap<(String, String), Audience>,
127}
128
129#[derive(Default)]
130pub struct StoreAudienceTransitions {
131    by_routing_id: HashMap<String, (Audience, String)>,
132}
133
134#[derive(Debug)]
135pub(crate) struct InboundStoreChangesets {
136    pub mirror: Vec<u8>,
137    pub rows: Vec<u8>,
138}
139
140impl RoutingChanges {
141    pub fn empty() -> Self {
142        Self {
143            store_mirror: Vec::new(),
144            private_routes: BTreeMap::new(),
145            deleted_rows: BTreeMap::new(),
146        }
147    }
148}
149
150struct PartitionGroup {
151    control: Option<CirclePartitionControl>,
152    group: Changegroup,
153}
154
155#[cfg(test)]
156mod tests;
coven_database/gate/audience.rs

coven_database/gate/
audience.rs
