Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! `file` mode durability: the redb write-through shadow behind the
3 : //! in-memory maps, its tables and key layout.
4 :
5 : use ::redb::{Database, Durability, TableDefinition};
6 : use antares_store::Kind;
7 :
8 : // ---- `file` mode: redb write-through shadow --------------------------------
9 : //
10 : // redb is durability only — queries and the matcher keep running on the
11 : // in-memory maps. Every mutation commits to redb (Durability::Immediate,
12 : // fsync) INSIDE the store's write-critical section, so redb apply order is
13 : // exactly memory apply order, and the commit happens before the store call
14 : // returns — i.e. before the HTTP ack (commit-before-ack). Boot rebuilds
15 : // the maps from the file and refuses to start on a format mismatch.
16 : //
17 : // Table per resource family, named after the spec resource, snake_cased.
18 : // The v0 memory store keeps one temporal doc per entity, so `attr_instances`
19 : // has no separate table; entityMaps are TTL-ephemeral and not durable state.
20 : pub(super) const T_ENTITIES: TableDefinition<&[u8], &[u8]> = TableDefinition::new("entities");
21 : pub(super) const T_SUBSCRIPTIONS: TableDefinition<&[u8], &[u8]> =
22 : TableDefinition::new("subscriptions");
23 : pub(super) const T_CSOURCE_REGISTRATIONS: TableDefinition<&[u8], &[u8]> =
24 : TableDefinition::new("csource_registrations");
25 : pub(super) const T_CSOURCE_SUBSCRIPTIONS: TableDefinition<&[u8], &[u8]> =
26 : TableDefinition::new("csource_subscriptions");
27 : pub(super) const T_TEMPORAL_ENTITIES: TableDefinition<&[u8], &[u8]> =
28 : TableDefinition::new("temporal_entities");
29 : pub(super) const T_JSONLD_CONTEXTS: TableDefinition<&[u8], &[u8]> =
30 : TableDefinition::new("jsonld_contexts");
31 : pub(super) const T_SNAPSHOTS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("snapshots");
32 : pub(super) const T_ENTITY_MAP_DOCS: TableDefinition<&[u8], &[u8]> =
33 : TableDefinition::new("entity_map_docs");
34 : pub(super) const T_DIST_SUBS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("dist_subs");
35 : pub(super) const T_DEAD_LETTERS: TableDefinition<&[u8], &[u8]> =
36 : TableDefinition::new("dead_letters");
37 : pub(super) const T_VIEWS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("views");
38 : pub(super) const T_META: TableDefinition<&str, &str> = TableDefinition::new("meta");
39 : /// On-disk format version: bump on any key/value shape change; an older
40 : /// or newer file refuses to load rather than being misread as valid data.
41 : pub(super) const FORMAT_VERSION: &str = "1";
42 :
43 : /// A table of the shadow: document bytes under a `tenant \0 id` key.
44 : pub(super) type Table = TableDefinition<'static, &'static [u8], &'static [u8]>;
45 :
46 440915 : pub(super) fn table_for(kind: Kind) -> Table {
47 440915 : match kind {
48 178488 : Kind::Entity => T_ENTITIES,
49 6637 : Kind::Subscription => T_SUBSCRIPTIONS,
50 3796 : Kind::Registration => T_CSOURCE_REGISTRATIONS,
51 781 : Kind::CSourceSubscription => T_CSOURCE_SUBSCRIPTIONS,
52 245032 : Kind::Temporal => T_TEMPORAL_ENTITIES,
53 807 : Kind::Snapshot => T_SNAPSHOTS,
54 2598 : Kind::EntityMap => T_ENTITY_MAP_DOCS,
55 2486 : Kind::DistSub => T_DIST_SUBS,
56 176 : Kind::DeadLetter => T_DEAD_LETTERS,
57 114 : Kind::View => T_VIEWS,
58 : }
59 440915 : }
60 :
61 : /// Key = `tenant \0 id`. Unambiguous: TenantId is `[A-Za-z0-9_-]{1,64}`
62 : /// by construction, so it can never contain the separator. Takes the tenant
63 : /// as the plain string the maps are keyed by, so a persisted removal can
64 : /// never be skipped for want of a re-parse.
65 443231 : pub(super) fn key_bytes(tenant: &str, id: &str) -> Vec<u8> {
66 443231 : let mut k = Vec::with_capacity(tenant.len().saturating_add(1).saturating_add(id.len()));
67 443231 : k.extend_from_slice(tenant.as_bytes());
68 443231 : k.push(0);
69 443231 : k.extend_from_slice(id.as_bytes());
70 443231 : k
71 443231 : }
72 :
73 1710 : pub(super) fn split_key(key: &[u8]) -> Option<(String, String)> {
74 4238 : let pos = key.iter().position(|&b| b == 0)?;
75 1710 : let (tenant, rest) = key.split_at_checked(pos)?;
76 : Some((
77 1710 : String::from_utf8(tenant.to_vec()).ok()?,
78 1710 : String::from_utf8(rest.get(1..)?.to_vec()).ok()?,
79 : ))
80 1710 : }
81 :
82 : pub(super) struct Shadow {
83 : pub(super) db: Database,
84 : }
85 :
86 : impl Shadow {
87 : /// One txn per mutation, fsynced before return. A failed commit
88 : /// aborts the process: the alternative is acking writes the file does not
89 : /// hold, which is the one lie a durable store must never tell.
90 : /// (Deliberately abort-on-commit-failure; per-request error plumbing only
91 : /// if a recoverable commit failure mode ever shows up in practice.)
92 12381 : pub(super) fn write(&self, table: Table, key: &[u8], value: Option<&[u8]>) {
93 12381 : self.write_all([(table, key, value)]);
94 12381 : }
95 :
96 : /// Several writes (`None` = remove the key) in ONE txn, one fsync: a
97 : /// multi-document operation (a purge, the sweep, a retention batch) is
98 : /// on disk whole or not at all. Same abort rule as [`Self::write`].
99 13013 : pub(super) fn write_all<'a>(
100 13013 : &self,
101 13013 : writes: impl IntoIterator<Item = (Table, &'a [u8], Option<&'a [u8]>)>,
102 13013 : ) {
103 13013 : let mut writes = writes.into_iter().peekable();
104 13013 : if writes.peek().is_none() {
105 475 : return;
106 12538 : }
107 12538 : let result = (|| -> Result<(), String> {
108 12538 : let mut tx = self.db.begin_write().map_err(|e| e.to_string())?;
109 12538 : tx.set_durability(Durability::Immediate)
110 12538 : .map_err(|e| e.to_string())?;
111 14197 : for (table, key, value) in writes {
112 14197 : let mut t = tx.open_table(table).map_err(|e| e.to_string())?;
113 14197 : match value {
114 9792 : Some(v) => {
115 9792 : t.insert(key, v).map_err(|e| e.to_string())?;
116 : }
117 : None => {
118 4405 : t.remove(key).map_err(|e| e.to_string())?;
119 : }
120 : }
121 : }
122 12538 : tx.commit().map_err(|e| e.to_string())
123 : })();
124 12538 : if let Err(e) = result {
125 0 : tracing::error!("redb commit failed: {e} — aborting: an acked write must be durable");
126 0 : std::process::abort();
127 12538 : }
128 13013 : }
129 : }
|
