fp-runtime 0.1.0

Runtime policies, evidence ledgers, and asupersync interop for the frankenpandas execution layer.
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
88
89
90
91
92
93
94
95
96
97
98
99
100

use std::{
    collections::BTreeMap,
    sync::{Arc, Mutex},
};

use serde::{Deserialize, Serialize};

use crate::asupersync::{
    codec::EncodedArtifact,
    config::{AsupersyncConfig, CapabilitySet, CxCapability},
    error::AsupersyncError,
    validate_capability_gate,
};

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TransferStatus {
    Completed,
    RetryableFailure,
    PermanentFailure,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TransferReport {
    pub artifact_id: String,
    pub bytes_transferred: usize,
    pub status: TransferStatus,
    pub detail: String,
}

pub trait TransportLayer {
    fn send(
        &self,
        artifact: EncodedArtifact,
        config: &AsupersyncConfig,
    ) -> Result<TransferReport, AsupersyncError>;

    fn receive(
        &self,
        artifact_id: &str,
        config: &AsupersyncConfig,
    ) -> Result<EncodedArtifact, AsupersyncError>;

    fn required_capabilities(&self) -> CapabilitySet {
        CapabilitySet::for_capability(CxCapability::Io)
            .union(CapabilitySet::for_capability(CxCapability::Remote))
    }
}

#[derive(Debug, Clone, Default)]
pub struct InMemoryTransport {
    storage: Arc<Mutex<BTreeMap<String, EncodedArtifact>>>,
}

impl InMemoryTransport {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }
}

impl TransportLayer for InMemoryTransport {
    fn send(
        &self,
        artifact: EncodedArtifact,
        config: &AsupersyncConfig,
    ) -> Result<TransferReport, AsupersyncError> {
        validate_capability_gate(config, self.required_capabilities())?;

        let mut guard = self.storage.lock().map_err(|_| {
            AsupersyncError::Transport("in-memory transport lock poisoned".to_string())
        })?;
        let bytes_transferred = artifact.encoded_bytes.len();
        let artifact_id = artifact.artifact_id.clone();
        guard.insert(artifact_id.clone(), artifact);

        Ok(TransferReport {
            artifact_id,
            bytes_transferred,
            status: TransferStatus::Completed,
            detail: "stored in in-memory transport".to_string(),
        })
    }

    fn receive(
        &self,
        artifact_id: &str,
        config: &AsupersyncConfig,
    ) -> Result<EncodedArtifact, AsupersyncError> {
        validate_capability_gate(config, self.required_capabilities())?;

        let guard = self.storage.lock().map_err(|_| {
            AsupersyncError::Transport("in-memory transport lock poisoned".to_string())
        })?;
        guard
            .get(artifact_id)
            .cloned()
            .ok_or_else(|| AsupersyncError::ArtifactNotFound(artifact_id.to_string()))
    }
}