community: one writer at a time per ledger, and an inbox publish that bumps from what the network really holds
This commit is contained in:
1 parent
0846b39d45
commit
733e3b57a8
3 files changed
+129
-5
No files matched your search
+53
-5
@@ -30,6 +30,7 @@ use persona_state::inbox as ib;
|
|||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use tauri::{Emitter, Manager};
|
use tauri::{Emitter, Manager};
|
||||||
|
|
||||||
|
use crate::ledger;
|
||||||
use crate::persona;
|
use crate::persona;
|
||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
|
|
||||||
@@ -147,17 +148,20 @@ fn dir(app: &tauri::AppHandle) -> Result<PathBuf, String> {
|
|||||||
.map_err(|e| e.to_string())?
|
.map_err(|e| e.to_string())?
|
||||||
.join("freenet"))
|
.join("freenet"))
|
||||||
}
|
}
|
||||||
|
/// Both the file the ledger lives in and the key its write lock is taken on.
|
||||||
|
const LEDGER_FILE: &str = "inbox-ledger.json";
|
||||||
|
|
||||||
fn load(app: &tauri::AppHandle) -> Ledger {
|
fn load(app: &tauri::AppHandle) -> Ledger {
|
||||||
dir(app)
|
dir(app)
|
||||||
.ok()
|
.ok()
|
||||||
.and_then(|d| std::fs::read(d.join("inbox-ledger.json")).ok())
|
.and_then(|d| std::fs::read(d.join(LEDGER_FILE)).ok())
|
||||||
.and_then(|b| serde_json::from_slice(&b).ok())
|
.and_then(|b| serde_json::from_slice(&b).ok())
|
||||||
.unwrap_or_default()
|
.unwrap_or_default()
|
||||||
}
|
}
|
||||||
fn save(app: &tauri::AppHandle, l: &Ledger) -> Result<(), String> {
|
fn save(app: &tauri::AppHandle, l: &Ledger) -> Result<(), String> {
|
||||||
let d = dir(app)?;
|
let d = dir(app)?;
|
||||||
std::fs::create_dir_all(&d).map_err(|e| e.to_string())?;
|
std::fs::create_dir_all(&d).map_err(|e| e.to_string())?;
|
||||||
std::fs::write(d.join("inbox-ledger.json"), serde_json::to_vec_pretty(l).map_err(|e| e.to_string())?)
|
std::fs::write(d.join(LEDGER_FILE), serde_json::to_vec_pretty(l).map_err(|e| e.to_string())?)
|
||||||
.map_err(|e| e.to_string())
|
.map_err(|e| e.to_string())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -204,13 +208,37 @@ pub fn status(app: &tauri::AppHandle) -> InboxStatus {
|
|||||||
/// Publish (or re-publish) how to reach this persona. Minting the box key
|
/// Publish (or re-publish) how to reach this persona. Minting the box key
|
||||||
/// and publishing the descriptor is what makes the inbox exist.
|
/// and publishing the descriptor is what makes the inbox exist.
|
||||||
pub async fn publish(app: &tauri::AppHandle) -> Result<InboxStatus, String> {
|
pub async fn publish(app: &tauri::AppHandle) -> Result<InboxStatus, String> {
|
||||||
|
let _ledger = ledger::hold(LEDGER_FILE).await;
|
||||||
let adapter = app.state::<AppState>().freenet.clone();
|
let adapter = app.state::<AppState>().freenet.clone();
|
||||||
let owner = own_key(app).await?;
|
let owner = own_key(app).await?;
|
||||||
|
// Whether we already had a box key, read before minting one.
|
||||||
|
let had_key = box_secret(app, false)?.is_some();
|
||||||
let secret = box_secret(app, true)?.expect("minted");
|
let secret = box_secret(app, true)?.expect("minted");
|
||||||
let mut l = load(app);
|
let mut l = load(app);
|
||||||
let first = l.desc_seq == 0;
|
// A box key with no sequence is the signature of a ledger that lost a
|
||||||
let seq = l.desc_seq + 1;
|
// write: the key is minted inside this function, just before the put. Ask
|
||||||
let key_id = if l.box_key_id == 0 { 1 } else { l.box_key_id };
|
// the network what it holds rather than trusting the zero, or we would
|
||||||
|
// re-put an existing contract at a sequence it already has and the
|
||||||
|
// descriptor would never move again.
|
||||||
|
let published = if l.desc_seq == 0 && had_key {
|
||||||
|
published_descriptor(app, &owner).await?
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
};
|
||||||
|
if let Some(p) = &published {
|
||||||
|
log::warn!(
|
||||||
|
"inbox: ledger said never published but the network holds sequence {} — bumping from there",
|
||||||
|
p.sequence
|
||||||
|
);
|
||||||
|
}
|
||||||
|
let net_seq = published.as_ref().map(|p| p.sequence).unwrap_or(0);
|
||||||
|
let seq = l.desc_seq.max(net_seq) + 1;
|
||||||
|
let first = seq == 1;
|
||||||
|
let key_id = match (l.box_key_id, published.as_ref().map(|p| p.box_key_id)) {
|
||||||
|
(0, Some(id)) if id > 0 => id,
|
||||||
|
(0, _) => 1,
|
||||||
|
(id, _) => id,
|
||||||
|
};
|
||||||
let mut d = ib::InboxDescriptor {
|
let mut d = ib::InboxDescriptor {
|
||||||
schema_version: ib::SCHEMA_VERSION,
|
schema_version: ib::SCHEMA_VERSION,
|
||||||
owner,
|
owner,
|
||||||
@@ -280,6 +308,25 @@ pub async fn resolve(app: &tauri::AppHandle, key: &[u8; 32]) -> Option<ib::Inbox
|
|||||||
desc
|
desc
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Our own descriptor exactly as the network holds it. Deliberately skips the
|
||||||
|
/// resolve cache: the whole point is to catch a ledger that lost a write.
|
||||||
|
async fn published_descriptor(
|
||||||
|
app: &tauri::AppHandle,
|
||||||
|
owner: &[u8; 32],
|
||||||
|
) -> Result<Option<ib::InboxDescriptor>, String> {
|
||||||
|
let adapter = app.state::<AppState>().freenet.clone();
|
||||||
|
let ids = FreenetAdapter::contract_ids(DESC_WASM.to_vec(), &ib::make_desc_params(owner));
|
||||||
|
match adapter.contract_get(&ids.instance, false).await {
|
||||||
|
Ok(bytes) => Ok(ib::validate_descriptor(owner, &bytes)),
|
||||||
|
// The contract genuinely is not out there: this really is a first put.
|
||||||
|
Err(e) if e.to_string().contains("not found") => Ok(None),
|
||||||
|
// Anything else means we could not ask. Refusing is the module rule —
|
||||||
|
// guessing "nothing" here re-puts an existing descriptor at a sequence
|
||||||
|
// the network already holds, and it would never move again.
|
||||||
|
Err(e) => Err(format!("could not read the published descriptor: {e}")),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// True when we can reach `key` through Freenet right now.
|
/// True when we can reach `key` through Freenet right now.
|
||||||
pub async fn reachable(app: &tauri::AppHandle, key: &[u8; 32]) -> bool {
|
pub async fn reachable(app: &tauri::AppHandle, key: &[u8; 32]) -> bool {
|
||||||
resolve(app, key).await.is_some()
|
resolve(app, key).await.is_some()
|
||||||
@@ -340,6 +387,7 @@ pub struct CheckResult {
|
|||||||
|
|
||||||
/// Read our inbox, open what we can, act on it, prune what we handled.
|
/// Read our inbox, open what we can, act on it, prune what we handled.
|
||||||
pub async fn check(app: &tauri::AppHandle) -> Result<CheckResult, String> {
|
pub async fn check(app: &tauri::AppHandle) -> Result<CheckResult, String> {
|
||||||
|
let _ledger = ledger::hold(LEDGER_FILE).await;
|
||||||
let adapter = app.state::<AppState>().freenet.clone();
|
let adapter = app.state::<AppState>().freenet.clone();
|
||||||
let owner = own_key(app).await?;
|
let owner = own_key(app).await?;
|
||||||
let Some(secret) = box_secret(app, false)? else {
|
let Some(secret) = box_secret(app, false)? else {
|
||||||
|
|||||||
@@ -0,0 +1,75 @@
|
|||||||
|
//! One writer at a time, per ledger file.
|
||||||
|
//!
|
||||||
|
//! Every Freenet slice keeps a small JSON ledger beside the node's data and
|
||||||
|
//! mutates it in the same shape: load → talk to the network → save. The
|
||||||
|
//! network leg takes seconds, so two overlapping operations both read the old
|
||||||
|
//! file and whichever finishes last writes its stale copy over the other's
|
||||||
|
//! result.
|
||||||
|
//!
|
||||||
|
//! That is what zeroed the inbox descriptor on 2026-09-02. The Community view
|
||||||
|
//! checks the inbox every 90 s; one of those checks loaded `desc_seq: 0` while
|
||||||
|
//! a publish was still putting the descriptor contract, and saved that stale
|
||||||
|
//! zero two minutes later — silently downgrading every sealed delivery to the
|
||||||
|
//! Soulseek fallback.
|
||||||
|
//!
|
||||||
|
//! `hold(file)` serializes a whole load → save span against any other span on
|
||||||
|
//! the same file. Locks are keyed by file name, so a slice may take another
|
||||||
|
//! slice's ledger while holding its own — the inbox does exactly that when it
|
||||||
|
//! dispatches sealed cargo into friends, shares and Spaces. The one rule: a
|
||||||
|
//! path must never re-take the ledger it already holds.
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::sync::{Arc, Mutex, OnceLock};
|
||||||
|
use tokio::sync::{Mutex as AsyncMutex, OwnedMutexGuard};
|
||||||
|
|
||||||
|
static LOCKS: OnceLock<Mutex<HashMap<&'static str, Arc<AsyncMutex<()>>>>> = OnceLock::new();
|
||||||
|
|
||||||
|
/// Take the write lock for one ledger file. Hold the guard across the whole
|
||||||
|
/// load → mutate → save span, network legs included.
|
||||||
|
pub async fn hold(file: &'static str) -> OwnedMutexGuard<()> {
|
||||||
|
let lock = {
|
||||||
|
let mut map = LOCKS.get_or_init(Default::default).lock().unwrap();
|
||||||
|
map.entry(file).or_default().clone()
|
||||||
|
};
|
||||||
|
lock.lock_owned().await
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use std::sync::atomic::{AtomicU32, Ordering};
|
||||||
|
|
||||||
|
/// The lost update this module exists to prevent: two tasks read the same
|
||||||
|
/// counter, await, then write. Under the guard the second must observe the
|
||||||
|
/// first one's write.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_holder_sees_the_previous_writers_value() {
|
||||||
|
static CELL: AtomicU32 = AtomicU32::new(0);
|
||||||
|
let one = tokio::spawn(async {
|
||||||
|
let _g = hold("test-race.json").await;
|
||||||
|
let seen = CELL.load(Ordering::SeqCst);
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(40)).await;
|
||||||
|
CELL.store(seen + 1, Ordering::SeqCst);
|
||||||
|
});
|
||||||
|
// Long enough that the first task is parked on its sleep, still holding.
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||||
|
let two = tokio::spawn(async {
|
||||||
|
let _g = hold("test-race.json").await;
|
||||||
|
let seen = CELL.load(Ordering::SeqCst);
|
||||||
|
CELL.store(seen + 1, Ordering::SeqCst);
|
||||||
|
});
|
||||||
|
one.await.unwrap();
|
||||||
|
two.await.unwrap();
|
||||||
|
assert_eq!(CELL.load(Ordering::SeqCst), 2, "the second write clobbered the first");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Different files must not block each other, or dispatching sealed cargo
|
||||||
|
/// out of a held inbox would deadlock.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn different_files_do_not_block() {
|
||||||
|
let _a = hold("test-a.json").await;
|
||||||
|
let _b = tokio::time::timeout(std::time::Duration::from_millis(500), hold("test-b.json"))
|
||||||
|
.await
|
||||||
|
.expect("a second ledger blocked on an unrelated one");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -9,6 +9,7 @@ mod persona;
|
|||||||
mod hero_crop;
|
mod hero_crop;
|
||||||
mod import;
|
mod import;
|
||||||
mod inbox;
|
mod inbox;
|
||||||
|
mod ledger;
|
||||||
mod hero_detector;
|
mod hero_detector;
|
||||||
mod loudness;
|
mod loudness;
|
||||||
mod lyric_align;
|
mod lyric_align;
|
||||||
|
|||||||
Reference in new issue
Block a user