the freenet layer: adapter, persona and profile services, the contracts bundled as commitments
This commit is contained in:
KeelrLeek committed
1 parent
63a0357651
commit
6dce4e2ac7
16 files changed
+3337
-10
No files matched your search
@@ -0,0 +1,789 @@
|
||||
//! The FreenetAdapter — SoulStream's ONLY door to a local Freenet node.
|
||||
//!
|
||||
//! Everything here follows the supplement design (§5.3) and the rules the
|
||||
//! F0–F7+F2b spikes recorded (research/freenet-*/RUN-RECORD.md):
|
||||
//!
|
||||
//! - **Off by default, kill-switchable.** `enable()` is the only thing that
|
||||
//! spawns work; `disable()` stops the supervisor, drops the connection,
|
||||
//! and ends every retry. A missing/stopped/incompatible node must never
|
||||
//! affect Soulseek or playback — nothing in here is on any audio path.
|
||||
//! - **The privilege boundary is this crate.** A loopback Freenet API is
|
||||
//! fully privileged and unauthenticated (F0): whatever can reach it owns
|
||||
//! the node. The webview therefore gets STATUS ONLY through the Tauri
|
||||
//! layer — never a node handle, never raw state bytes.
|
||||
//! - **Detection is cheap and three-valued** (absent / occupied-not-api /
|
||||
//! compatible), with its own short timeouts: a refused loopback connect
|
||||
//! costs ~2 s on Windows (F0), so the TCP probe gets 600 ms. A TYPED host
|
||||
//! error counts as compatible — local mode answers `ConnectedPeers` with
|
||||
//! "not supported" (F0), so refusals prove the protocol.
|
||||
//! - **One request in flight per connection.** The stdlib's `WebApi`
|
||||
//! correlates responses FIFO; the session serializes ops and routes
|
||||
//! unsolicited `UpdateNotification`s to a bounded event bus (lagging
|
||||
//! subscribers lose oldest events, honestly, rather than growing queues).
|
||||
//! - **Write rules from the spikes:** a node refuses the FIRST write to a
|
||||
//! contract it never held ("auto-fetch triggered", F5) — `update` retries
|
||||
//! over ~18 s and callers should prime with a GET (`get_then_update`);
|
||||
//! sequence-bearing states must be read-and-bumped by the caller (F5);
|
||||
//! a restarted node drops out of update propagation (F2) — services
|
||||
//! reconcile by relaying their own state after reconnect.
|
||||
//! - **Redacted diagnostics.** Logs carry ids, lengths and outcomes; never
|
||||
//! state bytes, never key material.
|
||||
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex as StdMutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use freenet_stdlib::client_api::{
|
||||
ClientRequest, ContractRequest, ContractResponse, HostResponse, NodeQuery, WebApi,
|
||||
};
|
||||
use freenet_stdlib::prelude::*;
|
||||
use serde::Serialize;
|
||||
use tokio::sync::{broadcast, mpsc, oneshot, watch};
|
||||
use tokio::time::timeout;
|
||||
|
||||
pub const DEFAULT_WS_PORT: u16 = 7509;
|
||||
|
||||
/// TCP probe budget — see the F0 note about Windows' 2 s refusal cost.
|
||||
const DETECT_TCP_TIMEOUT: Duration = Duration::from_millis(600);
|
||||
const DETECT_WS_TIMEOUT: Duration = Duration::from_secs(2);
|
||||
const DETECT_QUERY_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
const OP_TIMEOUT: Duration = Duration::from_secs(120);
|
||||
const PUT_TIMEOUT: Duration = Duration::from_secs(180);
|
||||
/// "auto-fetch triggered" retry schedule (F5): 6 attempts, 3 s apart.
|
||||
const AUTOFETCH_RETRIES: u32 = 6;
|
||||
const AUTOFETCH_BACKOFF: Duration = Duration::from_secs(3);
|
||||
/// Reconnect backoff while enabled but the node is missing/lost.
|
||||
const SUPERVISOR_BACKOFF: Duration = Duration::from_secs(5);
|
||||
const CMD_QUEUE: usize = 32;
|
||||
const EVENT_QUEUE: usize = 64;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
#[serde(tag = "kind", rename_all = "kebab-case")]
|
||||
pub enum AdapterStatus {
|
||||
/// The kill switch position. Nothing runs, nothing retries.
|
||||
Disabled,
|
||||
Detecting,
|
||||
/// Nothing listens on the port. NOT an error — Freenet is optional.
|
||||
NodeAbsent,
|
||||
/// Something listens but does not speak the client API.
|
||||
NodeOccupied,
|
||||
/// Speaks WebSocket but the protocol exchange failed undecipherably.
|
||||
NodeIncompatible { detail: String },
|
||||
Connected,
|
||||
/// Connection lost while enabled; the supervisor is retrying.
|
||||
Lost { detail: String },
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
#[serde(tag = "kind", rename_all = "kebab-case")]
|
||||
pub enum AdapterEvent {
|
||||
Status(AdapterStatus),
|
||||
/// A subscribed contract changed. Ids and lengths only — the webview
|
||||
/// never receives replicated bytes through the event path.
|
||||
ContractUpdate { instance: String, update_len: usize },
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum AdapterError {
|
||||
#[error("freenet layer is disabled")]
|
||||
Disabled,
|
||||
#[error("freenet node is not connected ({0:?})")]
|
||||
NotConnected(AdapterStatus),
|
||||
#[error("request timed out")]
|
||||
Timeout,
|
||||
#[error("adapter is shutting down")]
|
||||
ShuttingDown,
|
||||
#[error("node refused: {0}")]
|
||||
Node(String),
|
||||
#[error("{0}")]
|
||||
Other(String),
|
||||
}
|
||||
|
||||
type Result<T> = std::result::Result<T, AdapterError>;
|
||||
|
||||
/// What `contract_put`/`update` return so callers can address the instance
|
||||
/// later without recomputing.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct ContractIds {
|
||||
pub instance: String,
|
||||
pub code_hash: String,
|
||||
}
|
||||
|
||||
struct Cmd {
|
||||
req: ClientRequest<'static>,
|
||||
reply: oneshot::Sender<Result<HostResponse>>,
|
||||
op_timeout: Duration,
|
||||
}
|
||||
|
||||
struct Inner {
|
||||
ws_port: StdMutex<u16>,
|
||||
status: StdMutex<AdapterStatus>,
|
||||
events: broadcast::Sender<AdapterEvent>,
|
||||
/// Sender into the LIVE session, if any.
|
||||
session: StdMutex<Option<mpsc::Sender<Cmd>>>,
|
||||
/// Kill switch: flipping to false ends the supervisor and its session.
|
||||
enabled: watch::Sender<bool>,
|
||||
/// Distinguishes supervisor generations so a stale task can't clobber
|
||||
/// the status of a newer enable().
|
||||
generation: AtomicU64,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct FreenetAdapter {
|
||||
inner: Arc<Inner>,
|
||||
}
|
||||
|
||||
impl Default for FreenetAdapter {
|
||||
fn default() -> Self {
|
||||
let (enabled, _) = watch::channel(false);
|
||||
let (events, _) = broadcast::channel(EVENT_QUEUE);
|
||||
Self {
|
||||
inner: Arc::new(Inner {
|
||||
ws_port: StdMutex::new(DEFAULT_WS_PORT),
|
||||
status: StdMutex::new(AdapterStatus::Disabled),
|
||||
events,
|
||||
session: StdMutex::new(None),
|
||||
enabled,
|
||||
generation: AtomicU64::new(0),
|
||||
}),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn on the caller's tokio runtime when there is one (async commands,
|
||||
/// tests), else on a runtime the adapter owns. `enable()` is reached from
|
||||
/// Tauri's event-loop thread — the setup hook, sync commands — where a bare
|
||||
/// `tokio::spawn` is a non-unwinding panic that takes the whole process down
|
||||
/// (0xc0000409, seen live 2026-09-02 on the first toggle of the switch).
|
||||
fn spawn_anywhere<F>(fut: F)
|
||||
where
|
||||
F: std::future::Future<Output = ()> + Send + 'static,
|
||||
{
|
||||
match tokio::runtime::Handle::try_current() {
|
||||
Ok(handle) => {
|
||||
handle.spawn(fut);
|
||||
}
|
||||
Err(_) => {
|
||||
static RT: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
|
||||
let rt = RT.get_or_init(|| {
|
||||
tokio::runtime::Builder::new_multi_thread()
|
||||
.worker_threads(2)
|
||||
.thread_name("freenet-adapter")
|
||||
.enable_all()
|
||||
.build()
|
||||
.expect("freenet adapter runtime")
|
||||
});
|
||||
rt.spawn(fut);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
impl FreenetAdapter {
|
||||
pub fn set_ws_port(&self, port: u16) {
|
||||
*self.inner.ws_port.lock().unwrap() = port;
|
||||
}
|
||||
|
||||
pub fn ws_port(&self) -> u16 {
|
||||
*self.inner.ws_port.lock().unwrap()
|
||||
}
|
||||
|
||||
pub fn status(&self) -> AdapterStatus {
|
||||
self.inner.status.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
pub fn is_enabled(&self) -> bool {
|
||||
*self.inner.enabled.borrow()
|
||||
}
|
||||
|
||||
pub fn subscribe(&self) -> broadcast::Receiver<AdapterEvent> {
|
||||
self.inner.events.subscribe()
|
||||
}
|
||||
|
||||
fn set_status(&self, generation: u64, s: AdapterStatus) {
|
||||
if self.inner.generation.load(Ordering::SeqCst) != generation {
|
||||
return; // a newer enable()/disable() owns the status now
|
||||
}
|
||||
*self.inner.status.lock().unwrap() = s.clone();
|
||||
let _ = self.inner.events.send(AdapterEvent::Status(s));
|
||||
}
|
||||
|
||||
/// Turn the layer on: spawns the supervisor, which detects, connects,
|
||||
/// and reconnects with backoff until `disable()`.
|
||||
pub fn enable(&self) {
|
||||
if self.is_enabled() {
|
||||
return;
|
||||
}
|
||||
let generation = self.inner.generation.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
self.inner.enabled.send_replace(true);
|
||||
let this = self.clone();
|
||||
spawn_anywhere(async move {
|
||||
this.supervisor(generation).await;
|
||||
});
|
||||
log::info!("freenet adapter enabled (gen {generation})");
|
||||
}
|
||||
|
||||
/// The kill switch. Ends the supervisor, drops the session, stops every
|
||||
/// retry. Callers see `Disabled` errors immediately.
|
||||
pub fn disable(&self) {
|
||||
let generation = self.inner.generation.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
self.inner.enabled.send_replace(false);
|
||||
*self.inner.session.lock().unwrap() = None;
|
||||
*self.inner.status.lock().unwrap() = AdapterStatus::Disabled;
|
||||
let _ = self.inner.events.send(AdapterEvent::Status(AdapterStatus::Disabled));
|
||||
log::info!("freenet adapter disabled (gen {generation})");
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ detect
|
||||
|
||||
/// Three-valued node detection. Usable while disabled (it opens its own
|
||||
/// short-lived connection and touches no adapter state).
|
||||
pub async fn detect(&self) -> AdapterStatus {
|
||||
let port = self.ws_port();
|
||||
let addr = format!("127.0.0.1:{port}");
|
||||
match timeout(DETECT_TCP_TIMEOUT, tokio::net::TcpStream::connect(&addr)).await {
|
||||
Err(_) | Ok(Err(_)) => return AdapterStatus::NodeAbsent,
|
||||
Ok(Ok(stream)) => drop(stream),
|
||||
}
|
||||
let mut api = match timeout(DETECT_WS_TIMEOUT, connect_ws(port)).await {
|
||||
Err(_) => return AdapterStatus::NodeOccupied,
|
||||
Ok(Err(_)) => return AdapterStatus::NodeOccupied,
|
||||
Ok(Ok(api)) => api,
|
||||
};
|
||||
if api
|
||||
.send(ClientRequest::NodeQueries(NodeQuery::ConnectedPeers))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
return AdapterStatus::NodeIncompatible { detail: "send failed".into() };
|
||||
}
|
||||
match timeout(DETECT_QUERY_TIMEOUT, api.recv()).await {
|
||||
Ok(Ok(_)) => AdapterStatus::Connected,
|
||||
// A decoded, typed refusal IS protocol proof (local mode answers
|
||||
// ConnectedPeers with "not supported" — F0).
|
||||
Ok(Err(e)) if e.to_string().contains("not supported") => AdapterStatus::Connected,
|
||||
Ok(Err(e)) => AdapterStatus::NodeIncompatible { detail: truncate(&e.to_string(), 120) },
|
||||
Err(_) => AdapterStatus::NodeIncompatible { detail: "no reply to health query".into() },
|
||||
}
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ supervisor
|
||||
|
||||
async fn supervisor(&self, generation: u64) {
|
||||
let mut enabled_rx = self.inner.enabled.subscribe();
|
||||
loop {
|
||||
if self.inner.generation.load(Ordering::SeqCst) != generation || !*enabled_rx.borrow() {
|
||||
return;
|
||||
}
|
||||
self.set_status(generation, AdapterStatus::Detecting);
|
||||
let detected = self.detect().await;
|
||||
match detected {
|
||||
AdapterStatus::Connected => {}
|
||||
other => {
|
||||
self.set_status(generation, other);
|
||||
if wait_or_disabled(&mut enabled_rx, SUPERVISOR_BACKOFF).await {
|
||||
return;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
let port = self.ws_port();
|
||||
let api = match timeout(DETECT_WS_TIMEOUT, connect_ws(port)).await {
|
||||
Ok(Ok(api)) => api,
|
||||
_ => {
|
||||
self.set_status(generation, AdapterStatus::Lost { detail: "connect failed".into() });
|
||||
if wait_or_disabled(&mut enabled_rx, SUPERVISOR_BACKOFF).await {
|
||||
return;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let (tx, rx) = mpsc::channel::<Cmd>(CMD_QUEUE);
|
||||
*self.inner.session.lock().unwrap() = Some(tx);
|
||||
self.set_status(generation, AdapterStatus::Connected);
|
||||
log::info!("freenet adapter connected to 127.0.0.1:{port}");
|
||||
let reason = self.run_session(api, rx, &mut enabled_rx).await;
|
||||
*self.inner.session.lock().unwrap() = None;
|
||||
if self.inner.generation.load(Ordering::SeqCst) != generation || !*enabled_rx.borrow() {
|
||||
return;
|
||||
}
|
||||
self.set_status(generation, AdapterStatus::Lost { detail: truncate(&reason, 120) });
|
||||
log::warn!("freenet session ended: {}", truncate(&reason, 120));
|
||||
if wait_or_disabled(&mut enabled_rx, SUPERVISOR_BACKOFF).await {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One connection's lifetime. STRICTLY one request in flight (the wire
|
||||
/// correlates FIFO); notifications stream out between and during ops.
|
||||
async fn run_session(
|
||||
&self,
|
||||
mut api: WebApi,
|
||||
mut rx: mpsc::Receiver<Cmd>,
|
||||
enabled_rx: &mut watch::Receiver<bool>,
|
||||
) -> String {
|
||||
loop {
|
||||
tokio::select! {
|
||||
changed = enabled_rx.changed() => {
|
||||
if changed.is_err() || !*enabled_rx.borrow() {
|
||||
return "disabled".into();
|
||||
}
|
||||
}
|
||||
maybe_cmd = rx.recv() => {
|
||||
let Some(cmd) = maybe_cmd else { return "handle dropped".into() };
|
||||
if api.send(cmd.req).await.is_err() {
|
||||
let _ = cmd.reply.send(Err(AdapterError::Node("send failed".into())));
|
||||
return "send failed".into();
|
||||
}
|
||||
// Await THIS op's response; unsolicited notifications
|
||||
// arriving in between are forwarded, not misdelivered.
|
||||
let deadline = tokio::time::Instant::now() + cmd.op_timeout;
|
||||
loop {
|
||||
let left = deadline.saturating_duration_since(tokio::time::Instant::now());
|
||||
if left.is_zero() {
|
||||
let _ = cmd.reply.send(Err(AdapterError::Timeout));
|
||||
break;
|
||||
}
|
||||
match timeout(left, api.recv()).await {
|
||||
Err(_) => {
|
||||
let _ = cmd.reply.send(Err(AdapterError::Timeout));
|
||||
break;
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
let _ = cmd.reply.send(Err(AdapterError::Node(e.to_string())));
|
||||
break;
|
||||
}
|
||||
Ok(Ok(HostResponse::ContractResponse(
|
||||
ContractResponse::UpdateNotification { key, update },
|
||||
))) => {
|
||||
self.forward_notification(&key, &update);
|
||||
continue;
|
||||
}
|
||||
Ok(Ok(resp)) => {
|
||||
let _ = cmd.reply.send(Ok(resp));
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
streamed = api.recv() => {
|
||||
// Idle traffic: subscription notifications, or the end
|
||||
// of the connection.
|
||||
match streamed {
|
||||
Ok(HostResponse::ContractResponse(
|
||||
ContractResponse::UpdateNotification { key, update },
|
||||
)) => self.forward_notification(&key, &update),
|
||||
Ok(_) => { /* stray late response; correlation is FIFO, drop */ }
|
||||
Err(e) => return e.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn forward_notification(&self, key: &ContractKey, update: &UpdateData<'static>) {
|
||||
let len = match update {
|
||||
UpdateData::State(s) => s.as_ref().len(),
|
||||
UpdateData::Delta(d) => d.as_ref().len(),
|
||||
UpdateData::StateAndDelta { state, delta } => state.as_ref().len() + delta.as_ref().len(),
|
||||
_ => 0,
|
||||
};
|
||||
// Bounded broadcast: lagging receivers lose oldest events. Honest.
|
||||
let _ = self.inner.events.send(AdapterEvent::ContractUpdate {
|
||||
instance: key.id().encode(),
|
||||
update_len: len,
|
||||
});
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ ops
|
||||
|
||||
async fn request(&self, req: ClientRequest<'static>, op_timeout: Duration) -> Result<HostResponse> {
|
||||
if !self.is_enabled() {
|
||||
return Err(AdapterError::Disabled);
|
||||
}
|
||||
let tx = {
|
||||
let guard = self.inner.session.lock().unwrap();
|
||||
guard.clone()
|
||||
};
|
||||
let Some(tx) = tx else {
|
||||
return Err(AdapterError::NotConnected(self.status()));
|
||||
};
|
||||
let (reply_tx, reply_rx) = oneshot::channel();
|
||||
tx.send(Cmd { req, reply: reply_tx, op_timeout })
|
||||
.await
|
||||
.map_err(|_| AdapterError::ShuttingDown)?;
|
||||
reply_rx.await.map_err(|_| AdapterError::ShuttingDown)?
|
||||
}
|
||||
|
||||
fn container(code: Vec<u8>, params: &[u8]) -> (ContractContainer, ContractKey) {
|
||||
let params = Parameters::from(params.to_vec()).into_owned();
|
||||
let c = ContractContainer::Wasm(ContractWasmAPIVersion::V1(WrappedContract::new(
|
||||
Arc::new(ContractCode::from(code)),
|
||||
params,
|
||||
)));
|
||||
let k = c.key();
|
||||
(c, k)
|
||||
}
|
||||
|
||||
/// Compute the ids a (code, params) pair addresses — no network.
|
||||
pub fn contract_ids(code: Vec<u8>, params: &[u8]) -> ContractIds {
|
||||
let (_, key) = Self::container(code, params);
|
||||
ContractIds {
|
||||
instance: key.id().encode(),
|
||||
code_hash: key.encoded_code_hash(),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn contract_put(
|
||||
&self,
|
||||
code: Vec<u8>,
|
||||
params: &[u8],
|
||||
state: Vec<u8>,
|
||||
subscribe: bool,
|
||||
) -> Result<ContractIds> {
|
||||
let (container, key) = Self::container(code, params);
|
||||
let ids = ContractIds {
|
||||
instance: key.id().encode(),
|
||||
code_hash: key.encoded_code_hash(),
|
||||
};
|
||||
let resp = self
|
||||
.request(
|
||||
ContractRequest::Put {
|
||||
contract: container,
|
||||
state: WrappedState::new(state),
|
||||
related_contracts: RelatedContracts::new(),
|
||||
subscribe,
|
||||
blocking_subscribe: false,
|
||||
}
|
||||
.into(),
|
||||
PUT_TIMEOUT,
|
||||
)
|
||||
.await?;
|
||||
match resp {
|
||||
HostResponse::ContractResponse(ContractResponse::PutResponse { .. }) => Ok(ids),
|
||||
other => Err(unexpected("PutResponse", &other)),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn contract_get(
|
||||
&self,
|
||||
instance_b58: &str,
|
||||
subscribe: bool,
|
||||
) -> Result<Vec<u8>> {
|
||||
let id = parse_instance(instance_b58)?;
|
||||
let resp = self
|
||||
.request(
|
||||
ContractRequest::Get {
|
||||
key: id,
|
||||
return_contract_code: false,
|
||||
subscribe,
|
||||
blocking_subscribe: false,
|
||||
}
|
||||
.into(),
|
||||
OP_TIMEOUT,
|
||||
)
|
||||
.await
|
||||
// Depending on version, a never-put contract comes back either
|
||||
// as a typed NotFound or as an error frame "missing contract:
|
||||
// <id>" (observed on 0.2.132 local mode). Fold both into one
|
||||
// shape so services can match on "contract not found" alone.
|
||||
.map_err(|e| match e {
|
||||
AdapterError::Node(msg) if msg.contains("missing contract") => {
|
||||
AdapterError::Node("contract not found".into())
|
||||
}
|
||||
e => e,
|
||||
})?;
|
||||
match resp {
|
||||
HostResponse::ContractResponse(ContractResponse::GetResponse { state, .. }) => {
|
||||
Ok(state.to_vec())
|
||||
}
|
||||
HostResponse::ContractResponse(ContractResponse::NotFound { .. }) => {
|
||||
Err(AdapterError::Node("contract not found".into()))
|
||||
}
|
||||
other => Err(unexpected("GetResponse", &other)),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn contract_subscribe(&self, instance_b58: &str) -> Result<bool> {
|
||||
let id = parse_instance(instance_b58)?;
|
||||
let resp = self
|
||||
.request(
|
||||
ContractRequest::Subscribe { key: id, summary: None }.into(),
|
||||
OP_TIMEOUT,
|
||||
)
|
||||
.await?;
|
||||
match resp {
|
||||
HostResponse::ContractResponse(ContractResponse::SubscribeResponse {
|
||||
subscribed,
|
||||
..
|
||||
}) => Ok(subscribed),
|
||||
other => Err(unexpected("SubscribeResponse", &other)),
|
||||
}
|
||||
}
|
||||
|
||||
/// UPDATE with the recorded first-write behavior handled: a node that
|
||||
/// never held the contract answers "auto-fetch triggered" and asks the
|
||||
/// client to retry; we do, on the spike's schedule. Returns the
|
||||
/// post-merge summary bytes.
|
||||
pub async fn contract_update(
|
||||
&self,
|
||||
code: Vec<u8>,
|
||||
params: &[u8],
|
||||
delta: Vec<u8>,
|
||||
) -> Result<Vec<u8>> {
|
||||
let (_, key) = Self::container(code, params);
|
||||
let mut last: Option<AdapterError> = None;
|
||||
for attempt in 0..AUTOFETCH_RETRIES {
|
||||
if attempt > 0 {
|
||||
tokio::time::sleep(AUTOFETCH_BACKOFF).await;
|
||||
if !self.is_enabled() {
|
||||
return Err(AdapterError::Disabled); // kill switch ends retries
|
||||
}
|
||||
}
|
||||
let resp = self
|
||||
.request(
|
||||
ContractRequest::Update {
|
||||
key,
|
||||
data: UpdateData::Delta(StateDelta::from(delta.clone())),
|
||||
}
|
||||
.into(),
|
||||
OP_TIMEOUT,
|
||||
)
|
||||
.await;
|
||||
match resp {
|
||||
Ok(HostResponse::ContractResponse(ContractResponse::UpdateResponse {
|
||||
summary,
|
||||
..
|
||||
})) => return Ok(summary.as_ref().to_vec()),
|
||||
Ok(other) => return Err(unexpected("UpdateResponse", &other)),
|
||||
Err(AdapterError::Node(msg)) if msg.contains("auto-fetch triggered") => {
|
||||
log::info!(
|
||||
"freenet: first-write priming on attempt {attempt} (auto-fetch in flight)"
|
||||
);
|
||||
last = Some(AdapterError::Node(msg));
|
||||
}
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
}
|
||||
Err(last.unwrap_or(AdapterError::Timeout))
|
||||
}
|
||||
|
||||
/// The reliable first-write recipe from F5: one GET primes code+state on
|
||||
/// this node, then the update applies. Use for a contract this node may
|
||||
/// never have held.
|
||||
pub async fn get_then_update(
|
||||
&self,
|
||||
code: Vec<u8>,
|
||||
params: &[u8],
|
||||
delta: Vec<u8>,
|
||||
) -> Result<Vec<u8>> {
|
||||
let ids = Self::contract_ids(code.clone(), params);
|
||||
let _ = self.contract_get(&ids.instance, false).await; // prime; absence is fine
|
||||
self.contract_update(code, params, delta).await
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ delegates
|
||||
|
||||
fn delegate_container(code: Vec<u8>, params: &[u8]) -> (DelegateContainer, DelegateKey) {
|
||||
let params = Parameters::from(params.to_vec()).into_owned();
|
||||
let dc = DelegateCode::from(code);
|
||||
let d = Delegate::from((&dc, ¶ms));
|
||||
let c = DelegateContainer::Wasm(DelegateWasmAPIVersion::V1(d));
|
||||
let k = c.key().clone();
|
||||
(c, k)
|
||||
}
|
||||
|
||||
/// Compute the delegate key a (code, params) pair addresses — no
|
||||
/// network. Lets services pin and log their delegate generation.
|
||||
pub fn delegate_key_b58(code: Vec<u8>, params: &[u8]) -> String {
|
||||
let (_, key) = Self::delegate_container(code, params);
|
||||
bs58::encode(key.bytes()).into_string()
|
||||
}
|
||||
|
||||
/// Register (idempotent per node) and return the delegate key (b58).
|
||||
/// `cipher`/`nonce` are legacy fields the node ignores (core#4140).
|
||||
pub async fn delegate_register(&self, code: Vec<u8>, params: &[u8]) -> Result<String> {
|
||||
use freenet_stdlib::client_api::DelegateRequest;
|
||||
let (container, key) = Self::delegate_container(code, params);
|
||||
let resp = self
|
||||
.request(
|
||||
ClientRequest::DelegateOp(DelegateRequest::RegisterDelegate {
|
||||
delegate: container,
|
||||
cipher: [0u8; 32],
|
||||
nonce: [0u8; 24],
|
||||
}),
|
||||
OP_TIMEOUT,
|
||||
)
|
||||
.await?;
|
||||
match resp {
|
||||
HostResponse::DelegateResponse { .. } => Ok(bs58::encode(key.bytes()).into_string()),
|
||||
other => Err(unexpected("DelegateResponse", &other)),
|
||||
}
|
||||
}
|
||||
|
||||
/// One application message in, the delegate's app-message reply out.
|
||||
/// NOTE (F1): migration across delegate code generations must be
|
||||
/// verified by reading state back — `RegisterDelegateWithPredecessors`
|
||||
/// silently copies nothing in 0.2.132 (GHSA-824h-7x5x-wfmf).
|
||||
pub async fn delegate_message(
|
||||
&self,
|
||||
code: Vec<u8>,
|
||||
params: &[u8],
|
||||
payload: Vec<u8>,
|
||||
) -> Result<Vec<u8>> {
|
||||
use freenet_stdlib::client_api::DelegateRequest;
|
||||
let (_, key) = Self::delegate_container(code.clone(), params);
|
||||
let params_owned = Parameters::from(params.to_vec()).into_owned();
|
||||
let resp = self
|
||||
.request(
|
||||
ClientRequest::DelegateOp(DelegateRequest::ApplicationMessages {
|
||||
key,
|
||||
params: params_owned,
|
||||
inbound: vec![InboundDelegateMsg::ApplicationMessage(
|
||||
ApplicationMessage::new(payload),
|
||||
)],
|
||||
}),
|
||||
OP_TIMEOUT,
|
||||
)
|
||||
.await?;
|
||||
match resp {
|
||||
HostResponse::DelegateResponse { values, .. } => values
|
||||
.into_iter()
|
||||
.find_map(|v| match v {
|
||||
OutboundDelegateMsg::ApplicationMessage(m) => Some(m.payload),
|
||||
_ => None,
|
||||
})
|
||||
.ok_or_else(|| AdapterError::Node("no app message in delegate reply".into())),
|
||||
other => Err(unexpected("DelegateResponse", &other)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------ helpers
|
||||
|
||||
async fn connect_ws(port: u16) -> std::result::Result<WebApi, String> {
|
||||
let uri = format!("ws://127.0.0.1:{port}/v1/contract/command?encodingProtocol=native");
|
||||
let cfg = tokio_tungstenite::tungstenite::protocol::WebSocketConfig::default()
|
||||
.max_message_size(Some(100 * 1024 * 1024))
|
||||
.max_frame_size(Some(16 * 1024 * 1024));
|
||||
let (stream, _) = tokio_tungstenite::connect_async_with_config(&uri, Some(cfg), false)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(WebApi::start(stream))
|
||||
}
|
||||
|
||||
/// Wait out a backoff, returning true if the adapter was disabled meanwhile.
|
||||
async fn wait_or_disabled(enabled_rx: &mut watch::Receiver<bool>, dur: Duration) -> bool {
|
||||
tokio::select! {
|
||||
_ = tokio::time::sleep(dur) => !*enabled_rx.borrow(),
|
||||
changed = enabled_rx.changed() => changed.is_err() || !*enabled_rx.borrow(),
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_instance(b58: &str) -> Result<ContractInstanceId> {
|
||||
ContractInstanceId::try_from(b58.to_string())
|
||||
.map_err(|e| AdapterError::Other(format!("bad instance id: {e}")))
|
||||
}
|
||||
|
||||
fn unexpected(wanted: &str, got: &HostResponse) -> AdapterError {
|
||||
// Display, not Debug: HostResponse Display is summary-shaped and does
|
||||
// not dump state bytes.
|
||||
AdapterError::Node(format!("expected {wanted}, got {}", truncate(&got.to_string(), 120)))
|
||||
}
|
||||
|
||||
fn truncate(s: &str, n: usize) -> String {
|
||||
if s.len() <= n {
|
||||
return s.to_string();
|
||||
}
|
||||
let mut cut = n;
|
||||
while cut > 0 && !s.is_char_boundary(cut) {
|
||||
cut -= 1;
|
||||
}
|
||||
format!("{}…", &s[..cut])
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------ tests
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// The event-loop path: no tokio context at all. Must not panic, must
|
||||
/// flip enabled, and disable() must still end it.
|
||||
#[test]
|
||||
fn enable_works_outside_any_runtime() {
|
||||
std::thread::spawn(|| {
|
||||
assert!(tokio::runtime::Handle::try_current().is_err());
|
||||
let a = FreenetAdapter::default();
|
||||
a.enable();
|
||||
assert!(a.is_enabled());
|
||||
std::thread::sleep(std::time::Duration::from_millis(300));
|
||||
a.disable();
|
||||
assert!(!a.is_enabled());
|
||||
assert_eq!(a.status(), AdapterStatus::Disabled);
|
||||
})
|
||||
.join()
|
||||
.expect("no panic on the runtime-less path");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn disabled_is_the_default_and_ops_refuse() {
|
||||
let a = FreenetAdapter::default();
|
||||
assert_eq!(a.status(), AdapterStatus::Disabled);
|
||||
assert!(!a.is_enabled());
|
||||
let err = a.contract_get("11111111111111111111111111111111", false).await;
|
||||
assert!(matches!(err, Err(AdapterError::Disabled)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn detect_absent_on_closed_port() {
|
||||
let a = FreenetAdapter::default();
|
||||
a.set_ws_port(1); // nothing listens on port 1
|
||||
assert_eq!(a.detect().await, AdapterStatus::NodeAbsent);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn detect_occupied_on_non_api_listener() {
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let port = listener.local_addr().unwrap().port();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Ok((mut sock, _)) = listener.accept().await else { break };
|
||||
use tokio::io::AsyncWriteExt;
|
||||
// Answer like a confused HTTP server, then hang up.
|
||||
let _ = sock.write_all(b"HTTP/1.1 404 Not Found\r\n\r\n").await;
|
||||
}
|
||||
});
|
||||
let a = FreenetAdapter::default();
|
||||
a.set_ws_port(port);
|
||||
assert_eq!(a.detect().await, AdapterStatus::NodeOccupied);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn kill_switch_ends_everything() {
|
||||
let a = FreenetAdapter::default();
|
||||
a.set_ws_port(1);
|
||||
a.enable();
|
||||
assert!(a.is_enabled());
|
||||
// Give the supervisor a moment to enter its loop, then kill it.
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
a.disable();
|
||||
assert_eq!(a.status(), AdapterStatus::Disabled);
|
||||
assert!(!a.is_enabled());
|
||||
let err = a.contract_get("11111111111111111111111111111111", false).await;
|
||||
assert!(matches!(err, Err(AdapterError::Disabled)));
|
||||
// A stale supervisor tick must not resurrect a status.
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
assert_eq!(a.status(), AdapterStatus::Disabled);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn contract_ids_are_deterministic_offline() {
|
||||
let code = vec![0u8, 1, 2, 3];
|
||||
let a = FreenetAdapter::contract_ids(code.clone(), b"params");
|
||||
let b = FreenetAdapter::contract_ids(code, b"params");
|
||||
assert_eq!(a.instance, b.instance);
|
||||
assert_eq!(a.code_hash, b.code_hash);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user