kmp_adapter_embedded/adapter/
node_card.rs

1use kmp_domain::{
2    AuthorNodeCard, NodeCard, NodeCardStore, NodeCardWriteFuture, PortError, node_card_policy,
3};
4
5use super::engine::{Key, ReadTx, Table, WriteTx};
6use super::serdes::{CardRecord, NodeRecord, decode, encode};
7use super::store::EmbeddedKernelStore;
8
9/// Cards in requested order, preserving duplicates and missing slots.
10///
11/// Takes the caller's transaction, so a trace presents cards from the same
12/// snapshot it read descriptors and adjacency from. A card read against a
13/// later snapshot could describe a body this response never showed.
14pub(super) fn read_batch(
15    tx: &dyn ReadTx,
16    node_ids: &[String],
17    language: &str,
18) -> Result<Vec<Option<NodeCard>>, PortError> {
19    node_ids
20        .iter()
21        .map(|id| {
22            tx.get(Table::Cards, Key::Str2(id, language))?
23                .map(|raw| decode::<CardRecord>("node card", &raw).map(Into::into))
24                .transpose()
25        })
26        .collect()
27}
28
29fn stored_card(
30    tx: &dyn WriteTx,
31    node_id: &str,
32    language: &str,
33) -> Result<Option<NodeCard>, PortError> {
34    tx.get(Table::Cards, Key::Str2(node_id, language))?
35        .map(|raw| decode::<CardRecord>("node card", &raw).map(Into::into))
36        .transpose()
37}
38
39impl NodeCardStore for EmbeddedKernelStore {
40    fn author_node_card(&self, command: AuthorNodeCard) -> NodeCardWriteFuture<'_> {
41        Box::pin(async move {
42            self.run(move |store| {
43                // One write transaction for the whole decision. The node, its
44                // body descriptor and the stored card are read here, inside
45                // the lock the write already holds, so nothing can move
46                // between the check and the insert — which is the only thing
47                // that makes the declared source version mean anything.
48                let mut tx = store.begin_write()?;
49                let node = tx
50                    .get(Table::Nodes, Key::Str(&command.node_id))?
51                    .map(|raw| decode::<NodeRecord>("card node", &raw)?.into_projection())
52                    .transpose()?;
53                // The header, never the record. A reader already paid to read
54                // the body it is writing about; the kernel checking that write
55                // must not pay for it a second time.
56                let descriptor =
57                    super::node_body_descriptor::read_one(tx.as_ref(), &command.node_id)?;
58                let existing = stored_card(tx.as_ref(), &command.node_id, &command.language)?;
59
60                let admitted = match node_card_policy::admit(
61                    &command,
62                    node.as_ref(),
63                    descriptor.as_ref(),
64                    existing.as_ref(),
65                    None,
66                ) {
67                    Ok(card) => card,
68                    // Dropping the transaction discards it: a refused card
69                    // leaves the store exactly as it found it.
70                    Err(rejection) => return Ok(Err(rejection)),
71                };
72
73                let record = CardRecord::from(admitted.clone());
74                let encoded = encode("node card", &record)?;
75                tx.insert(
76                    Table::Cards,
77                    Key::Str2(&admitted.node_id, &admitted.language),
78                    &encoded,
79                )?;
80                tx.commit()?;
81                Ok(Ok(admitted))
82            })
83            .await
84        })
85    }
86}
87
88#[cfg(test)]
89#[path = "node_card_tests.rs"]
90mod tests;
kmp_adapter_embedded/adapter/node_card.rs

kmp_adapter_embedded/adapter/
node_card.rs
