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_META: TableDefinition<&str, &str> = TableDefinition::new("meta");
38 : /// On-disk format version: bump on any key/value shape change; an older
39 : /// or newer file refuses to load rather than being misread as valid data.
40 : pub(super) const FORMAT_VERSION: &str = "1";
41 :
42 56559 : pub(super) fn table_for(kind: Kind) -> TableDefinition<'static, &'static [u8], &'static [u8]> {
43 56559 : match kind {
44 37990 : Kind::Entity => T_ENTITIES,
45 1480 : Kind::Subscription => T_SUBSCRIPTIONS,
46 1877 : Kind::Registration => T_CSOURCE_REGISTRATIONS,
47 108 : Kind::CSourceSubscription => T_CSOURCE_SUBSCRIPTIONS,
48 10812 : Kind::Temporal => T_TEMPORAL_ENTITIES,
49 624 : Kind::Snapshot => T_SNAPSHOTS,
50 2510 : Kind::EntityMap => T_ENTITY_MAP_DOCS,
51 1008 : Kind::DistSub => T_DIST_SUBS,
52 150 : Kind::DeadLetter => T_DEAD_LETTERS,
53 : }
54 56559 : }
55 :
56 : /// Key = `tenant \0 id`. Unambiguous: TenantId is `[A-Za-z0-9_-]{1,64}`
57 : /// by construction, so it can never contain the separator. Takes the tenant
58 : /// as the plain string the maps are keyed by, so a persisted removal can
59 : /// never be skipped for want of a re-parse.
60 55869 : pub(super) fn key_bytes(tenant: &str, id: &str) -> Vec<u8> {
61 55869 : let mut k = Vec::with_capacity(tenant.len() + 1 + id.len());
62 55869 : k.extend_from_slice(tenant.as_bytes());
63 55869 : k.push(0);
64 55869 : k.extend_from_slice(id.as_bytes());
65 55869 : k
66 55869 : }
67 :
68 80 : pub(super) fn split_key(key: &[u8]) -> Option<(String, String)> {
69 840 : let pos = key.iter().position(|&b| b == 0)?;
70 : Some((
71 80 : String::from_utf8(key[..pos].to_vec()).ok()?,
72 80 : String::from_utf8(key[pos + 1..].to_vec()).ok()?,
73 : ))
74 80 : }
75 :
76 : pub(super) struct Shadow {
77 : pub(super) db: Database,
78 : }
79 :
80 : impl Shadow {
81 : /// One txn per mutation, fsynced before return. A failed commit
82 : /// aborts the process: the alternative is acking writes the file does not
83 : /// hold, which is the one lie a durable store must never tell.
84 : /// (Deliberately abort-on-commit-failure; per-request error plumbing only
85 : /// if a recoverable commit failure mode ever shows up in practice.)
86 1241 : pub(super) fn write(
87 1241 : &self,
88 1241 : table: TableDefinition<&[u8], &[u8]>,
89 1241 : key: &[u8],
90 1241 : value: Option<&[u8]>,
91 1241 : ) {
92 1241 : let result = (|| -> Result<(), String> {
93 1241 : let mut tx = self.db.begin_write().map_err(|e| e.to_string())?;
94 1241 : tx.set_durability(Durability::Immediate)
95 1241 : .map_err(|e| e.to_string())?;
96 : {
97 1241 : let mut t = tx.open_table(table).map_err(|e| e.to_string())?;
98 1241 : match value {
99 772 : Some(v) => {
100 772 : t.insert(key, v).map_err(|e| e.to_string())?;
101 : }
102 : None => {
103 469 : t.remove(key).map_err(|e| e.to_string())?;
104 : }
105 : }
106 : }
107 1241 : tx.commit().map_err(|e| e.to_string())
108 : })();
109 1241 : if let Err(e) = result {
110 0 : tracing::error!("redb commit failed: {e} — aborting: an acked write must be durable");
111 0 : std::process::abort();
112 1241 : }
113 1241 : }
114 : }
|
