sim_lib_bridge/
collab.rs

1use std::sync::Arc;
2
3use sim_codec_bridge::{
4    BridgeBook, BridgePacket, BridgePatchPayload, BridgeVotePayload, stamp_packet_cid,
5};
6use sim_kernel::{Cx, DefaultFactory, EagerPolicy, Error, Result, Symbol};
7
8use crate::parent::parents_contain_cid;
9use crate::rx_check;
10
11/// Declared policy for combining collaboration contributions.
12#[derive(Clone, Debug, PartialEq, Eq)]
13pub enum MergePolicy {
14    /// Accept one patch contribution.
15    Single,
16    /// Accept one patch contribution once enough vote records target it.
17    Quorum {
18        /// Minimum number of matching vote records.
19        min_votes: u32,
20    },
21    /// Accept a synthesizer patch contribution once enough vote records target it.
22    SynthesisThenVote {
23        /// Seat name of the synthesizer whose patch is eligible.
24        synthesizer: String,
25        /// Minimum number of matching vote records.
26        min_votes: u32,
27    },
28}
29
30/// Merge patch replies by exact parent content id and target path.
31///
32/// The selected packet is checked as a reply to `base`, so the root packet's
33/// `Return` contract still decides whether the merged payload is admissible.
34pub fn merge_bridge_replies(
35    base: &BridgePacket,
36    replies: &[BridgePacket],
37    policy: &MergePolicy,
38) -> Result<BridgePacket> {
39    let base_cid = base.header.cid.as_deref().ok_or_else(|| {
40        Error::Eval("BRIDGE collaboration merge requires a stamped base packet".to_owned())
41    })?;
42    let patches = patch_candidates(base_cid, replies)?;
43    if patches.is_empty() {
44        return Err(Error::Eval(
45            "BRIDGE collaboration merge found no patch for the base packet".to_owned(),
46        ));
47    }
48    require_one_target(&patches)?;
49    let selected = select_patch(base_cid, replies, &patches, policy)?;
50    let merged = stamp_packet_cid(&selected.packet.canonicalized())?;
51    check_merged_reply(base, &merged)?;
52    Ok(merged)
53}
54
55#[derive(Clone)]
56struct PatchCandidate<'a> {
57    packet: &'a BridgePacket,
58    patch: BridgePatchPayload,
59}
60
61fn patch_candidates<'a>(
62    base_cid: &str,
63    replies: &'a [BridgePacket],
64) -> Result<Vec<PatchCandidate<'a>>> {
65    let mut patches = Vec::new();
66    for packet in replies {
67        if !parents_contain_cid(&packet.header.parents, base_cid) {
68            continue;
69        }
70        for part in &packet.body {
71            if part.kind != Symbol::qualified("bridge", "Patch") {
72                continue;
73            }
74            let patch = BridgePatchPayload::from_expr(&part.payload)?;
75            if patch.parent_cid == base_cid {
76                patches.push(PatchCandidate { packet, patch });
77            }
78        }
79    }
80    Ok(patches)
81}
82
83fn require_one_target(patches: &[PatchCandidate<'_>]) -> Result<()> {
84    let target = patches[0].patch.target.as_str();
85    if patches
86        .iter()
87        .all(|candidate| candidate.patch.target == target)
88    {
89        return Ok(());
90    }
91    Err(Error::Eval(
92        "BRIDGE collaboration merge patches must share one exact target path".to_owned(),
93    ))
94}
95
96fn select_patch<'a>(
97    base_cid: &str,
98    replies: &'a [BridgePacket],
99    patches: &[PatchCandidate<'a>],
100    policy: &MergePolicy,
101) -> Result<PatchCandidate<'a>> {
102    match policy {
103        MergePolicy::Single => {
104            if patches.len() == 1 {
105                Ok(patches[0].clone())
106            } else {
107                Err(Error::Eval(
108                    "BRIDGE single merge requires exactly one patch".to_owned(),
109                ))
110            }
111        }
112        MergePolicy::Quorum { min_votes } => {
113            select_quorum_patch(base_cid, replies, patches, *min_votes)
114        }
115        MergePolicy::SynthesisThenVote {
116            synthesizer,
117            min_votes,
118        } => {
119            if synthesizer.is_empty() {
120                return Err(Error::Eval(
121                    "BRIDGE synthesis merge requires a synthesizer seat".to_owned(),
122                ));
123            }
124            let synthesized = patches
125                .iter()
126                .filter(|candidate| candidate.packet.header.from == *synthesizer)
127                .cloned()
128                .collect::<Vec<_>>();
129            select_quorum_patch(base_cid, replies, &synthesized, *min_votes)
130        }
131    }
132}
133
134fn select_quorum_patch<'a>(
135    base_cid: &str,
136    replies: &[BridgePacket],
137    patches: &[PatchCandidate<'a>],
138    min_votes: u32,
139) -> Result<PatchCandidate<'a>> {
140    if min_votes == 0 {
141        return Err(Error::Eval(
142            "BRIDGE quorum merge requires at least one vote".to_owned(),
143        ));
144    }
145    let eligible = patches
146        .iter()
147        .filter(|candidate| {
148            votes_for_target(base_cid, replies, &candidate.patch.target)
149                .map(|votes| votes >= min_votes)
150                .unwrap_or(false)
151        })
152        .cloned()
153        .collect::<Vec<_>>();
154    if eligible.len() == 1 {
155        Ok(eligible[0].clone())
156    } else if eligible.is_empty() {
157        Err(Error::Eval(
158            "BRIDGE quorum merge found no patch with enough votes".to_owned(),
159        ))
160    } else {
161        Err(Error::Eval(
162            "BRIDGE quorum merge selected more than one patch".to_owned(),
163        ))
164    }
165}
166
167fn votes_for_target(base_cid: &str, replies: &[BridgePacket], target: &str) -> Result<u32> {
168    let mut votes = 0u32;
169    for packet in replies {
170        if !parents_contain_cid(&packet.header.parents, base_cid) {
171            continue;
172        }
173        for part in &packet.body {
174            if part.kind != Symbol::qualified("bridge", "Vote") {
175                continue;
176            }
177            let vote = BridgeVotePayload::from_expr(&part.payload)?;
178            if vote.target == target {
179                votes += 1;
180            }
181        }
182    }
183    Ok(votes)
184}
185
186fn check_merged_reply(base: &BridgePacket, merged: &BridgePacket) -> Result<()> {
187    let book = BridgeBook::standard();
188    let mut cx = Cx::new(Arc::new(EagerPolicy), Arc::new(DefaultFactory));
189    let report = rx_check(&mut cx, &book, merged, Some(base))?;
190    if report.accepted() {
191        return Ok(());
192    }
193    Err(Error::Eval(format!(
194        "BRIDGE merged reply failed receive check: {:?}",
195        report.obligations
196    )))
197}
sim_lib_bridge/collab.rs

sim_lib_bridge/
collab.rs
