liminal-server 0.19.1

Standalone server for the liminal messaging bus
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

use super::handler::ProductionParticipantHandler;
use crate::server::participant::{
    ParticipantConnectionContext, ParticipantConnectionConversations, ParticipantSemanticError,
};
use liminal_protocol::wire::{
    BindingEpoch, RecordAdmission, RecordAdmissionResponse, ServerValue, SettlementAnswer,
    SettlementOutcome, SettlementQuery,
};

const fn answer(request: &SettlementQuery, outcome: SettlementOutcome) -> ServerValue {
    ServerValue::SettlementAnswer(SettlementAnswer {
        conversation_id: request.conversation_id,
        participant_id: request.participant_id,
        record_admission_attempt_token: request.record_admission_attempt_token,
        outcome,
    })
}

impl ProductionParticipantHandler {
    pub(super) fn answer_settlement_query(
        &self,
        context: ParticipantConnectionContext,
        conversations: &ParticipantConnectionConversations,
        request: &SettlementQuery,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        let Some(cell) = self.existing_cell(request.conversation_id)? else {
            return Ok(answer(request, SettlementOutcome::Unknown));
        };
        let owner = cell
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant settlement owner lock is poisoned".to_owned(),
            })?;
        let Some(authority) = owner.as_ref() else {
            drop(owner);
            return Ok(answer(request, SettlementOutcome::Unknown));
        };
        let facts = self.operation_facts(context, request.conversation_id, conversations)?;
        let admission = RecordAdmission {
            conversation_id: request.conversation_id,
            participant_id: request.participant_id,
            capability_generation: request.capability_generation,
            record_admission_attempt_token: request.record_admission_attempt_token,
            payload: Vec::new(),
        };
        let receiving_epoch =
            BindingEpoch::new(facts.receiving_incarnation, request.capability_generation);
        if let Some(outcome) =
            authority.classify_record_admission_authority(&admission, &facts, receiving_epoch)
        {
            drop(owner);
            return Ok(outcome.value);
        }
        let token = request.record_admission_attempt_token.into_bytes();
        let key = (token, request.participant_id, request.payload_fingerprint);
        if let Some(delivery_seq) = authority.committed_admissions.get(&key) {
            let value = answer(
                request,
                SettlementOutcome::Committed {
                    delivery_seq: *delivery_seq,
                },
            );
            drop(owner);
            return Ok(value);
        }
        if authority
            .committed_admissions
            .range(
                (token, request.participant_id, [0; 32])
                    ..=(token, request.participant_id, [0xff; 32]),
            )
            .next()
            .is_some()
        {
            drop(owner);
            return Ok(RecordAdmissionResponse::attempt_token_body_conflict(
                request.record_admission_attempt_token,
                request.conversation_id,
                request.participant_id,
                request.capability_generation,
            )
            .into_server_value());
        }
        drop(owner);
        Ok(answer(request, SettlementOutcome::NotCommitted))
    }
}