use anyhow::{anyhow, Result};
use futures::{FutureExt, StreamExt};
use iroh_docs::{engine::LiveEvent, protocol::Docs, NamespaceId};
use std::time::Duration;
fn ticket_retry_delay(attempt: usize, namespace: NamespaceId) -> Duration {
let base = 400u64
.saturating_mul(1u64 << attempt.saturating_sub(1).min(7))
.min(30_000);
let bytes: [u8; 32] = namespace.into();
let jitter = (u64::from(bytes[0]) + attempt as u64) % (base / 10 + 1);
Duration::from_millis(base.saturating_add(jitter).min(30_000))
}
pub(crate) fn spawn_ticket_sync_recovery(
docs: Docs,
namespace: NamespaceId,
peers: Vec<iroh::EndpointAddr>,
mut events: n0_future::boxed::BoxStream<Result<LiveEvent>>,
initial_failed: bool,
) -> TicketSyncTask {
let (cancel, cancelled) = async_channel::bounded::<()>(1);
let join = n0_future::task::spawn(async move {
let mut retry_pending = initial_failed;
let mut attempt = 0usize;
let mut synced_once = false;
let mut watchdog_at = n0_future::time::Instant::now() + Duration::from_secs(10);
loop {
if retry_pending {
attempt = attempt.saturating_add(1);
let delay = n0_future::time::sleep(ticket_retry_delay(attempt, namespace)).fuse();
let stopped = cancelled.recv().fuse();
futures::pin_mut!(delay, stopped);
futures::select! {
_ = delay => {},
_ = stopped => break,
}
let result = async {
let doc =
docs.api().open(namespace).await?.ok_or_else(|| {
anyhow!("imported iroh-docs namespace is unavailable")
})?;
let start_result = doc.start_sync(peers.clone()).await;
let close_result = doc.close().await;
start_result?;
close_result
}
.await;
if result.is_err() {
continue;
}
retry_pending = false;
watchdog_at = n0_future::time::Instant::now() + Duration::from_secs(10);
}
let event = if synced_once {
let next = events.next().fuse();
let stopped = cancelled.recv().fuse();
futures::pin_mut!(next, stopped);
futures::select! {
event = next => event,
_ = stopped => break,
}
} else {
let next = events.next().fuse();
let watchdog = n0_future::time::sleep_until(watchdog_at).fuse();
let stopped = cancelled.recv().fuse();
futures::pin_mut!(next, watchdog, stopped);
futures::select! {
event = next => event,
_ = watchdog => {
retry_pending = true;
continue;
},
_ = stopped => break,
}
};
match event {
Some(Ok(LiveEvent::SyncFinished(event)))
if event.result.is_ok() && peers.iter().any(|peer| peer.id == event.peer) =>
{
synced_once = true;
attempt = 0;
}
Some(Ok(LiveEvent::SyncFinished(event)))
if event.result.is_err() && peers.iter().any(|peer| peer.id == event.peer) =>
{
synced_once = false;
retry_pending = true;
}
Some(Ok(LiveEvent::NeighborDown(lost)))
if peers.iter().any(|peer| peer.id == lost) =>
{
synced_once = false;
retry_pending = true;
}
Some(Err(_)) => {
synced_once = false;
retry_pending = true;
}
Some(Ok(_)) => {}
None => break,
}
}
});
TicketSyncTask { cancel, join }
}
pub(crate) struct TicketSyncTask {
pub(crate) cancel: async_channel::Sender<()>,
pub(crate) join: n0_future::task::JoinHandle<()>,
}
#[cfg(all(test, not(target_arch = "wasm32")))]
mod tests {
use super::*;
use bytes::Bytes;
use iroh::{
endpoint::{presets, Connection},
protocol::{AcceptError, ProtocolHandler, Router},
Endpoint,
};
use iroh_blobs::{store::mem::MemStore, BlobsProtocol};
use iroh_docs::api::protocol::{AddrInfoOptions, ShareMode};
use iroh_gossip::net::Gossip;
use std::sync::{Arc, Mutex};
#[derive(Debug)]
struct ObservedGossip {
gossip: Gossip,
connections: Arc<Mutex<Vec<Connection>>>,
}
impl ProtocolHandler for ObservedGossip {
async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
self.connections.lock().unwrap().push(connection.clone());
self.gossip.accept(connection).await
}
}
async fn node() -> Result<(Docs, MemStore, Router, Arc<Mutex<Vec<Connection>>>)> {
let endpoint = Endpoint::builder(presets::N0)
.relay_mode(iroh::RelayMode::Disabled)
.bind()
.await?;
let blobs = MemStore::default();
let gossip = Gossip::builder().spawn(endpoint.clone());
let docs = Docs::memory()
.spawn(endpoint.clone(), (*blobs).clone(), gossip.clone())
.await?;
let connections = Arc::new(Mutex::new(Vec::new()));
let router = Router::builder(endpoint)
.accept(iroh_blobs::ALPN, BlobsProtocol::new(&blobs, None))
.accept(iroh_docs::ALPN, docs.clone())
.accept(
iroh_gossip::ALPN,
ObservedGossip {
gossip,
connections: connections.clone(),
},
)
.spawn();
Ok((docs, blobs, router, connections))
}
#[tokio::test]
async fn imported_ticket_recovers_writes_after_its_only_gossip_neighbor_closes() -> Result<()> {
let (host, host_blobs, host_router, host_connections) = node().await?;
let (client, _, client_router, client_connections) = node().await?;
let doc = host.create().await?;
let author = host.author_default().await?;
doc.set_bytes(
author,
Bytes::from_static(b"before"),
Bytes::from_static(b"initial"),
)
.await?;
let ticket = doc
.share(ShareMode::Write, AddrInfoOptions::RelayAndAddresses)
.await?;
let namespace = ticket.capability.id();
let peers = ticket.nodes.clone();
let replica = client.api().import_namespace(ticket.capability).await?;
let recovery_events = replica.subscribe().await?;
let mut observed = replica.subscribe().await?;
replica.start_sync(peers.clone()).await?;
let recovery = spawn_ticket_sync_recovery(
client.clone(),
namespace,
peers,
Box::pin(recovery_events),
false,
);
tokio::time::timeout(Duration::from_secs(15), async {
let (mut synced, mut neighbor) = (false, false);
while let Some(event) = observed.next().await {
match event? {
LiveEvent::SyncFinished(event) if event.result.is_ok() => synced = true,
LiveEvent::NeighborUp(_) => neighbor = true,
_ => {}
}
if synced && neighbor {
return Ok::<_, anyhow::Error>(());
}
}
anyhow::bail!("initial Docs stream ended")
})
.await??;
let connections: Vec<_> = host_connections
.lock()
.unwrap()
.drain(..)
.chain(client_connections.lock().unwrap().drain(..))
.collect();
assert!(
!connections.is_empty(),
"test must interrupt an established gossip edge"
);
for connection in connections {
connection.close(0u32.into(), b"carrier replacement");
}
tokio::time::timeout(Duration::from_secs(5), async {
while let Some(event) = observed.next().await {
if matches!(event?, LiveEvent::NeighborDown(_)) {
return Ok::<_, anyhow::Error>(());
}
}
anyhow::bail!("Docs stream ended before neighbor loss")
})
.await??;
let key = Bytes::from_static(b"after-disconnect");
let content = Bytes::from_static(b"written after the first successful sync");
let writer = client.author_default().await?;
let hash = replica
.set_bytes(writer, key.clone(), content.clone())
.await?;
let propagated = tokio::time::timeout(Duration::from_secs(8), async {
loop {
if let Some(entry) = doc.get_exact(writer, &key, false).await? {
if entry.content_hash() == hash {
if let Ok(bytes) = host_blobs.blobs().get_bytes(hash).await {
if bytes == content {
return Ok::<_, anyhow::Error>(());
}
}
}
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await;
if propagated.is_err() {
recovery.cancel.close();
recovery.join.await?;
client_router.shutdown().await?;
host_router.shutdown().await?;
anyhow::bail!("write did not propagate after the sole gossip neighbor closed");
}
propagated.unwrap()?;
let _ = tokio::time::timeout(Duration::from_secs(1), async {
while observed.next().await.is_some() {}
})
.await;
let extra_sync = tokio::time::timeout(Duration::from_secs(11), async {
while let Some(event) = observed.next().await {
if matches!(event?, LiveEvent::SyncFinished(_)) {
return Ok::<_, anyhow::Error>(());
}
}
anyhow::bail!("Docs stream ended during quiet window")
})
.await;
assert!(
extra_sync.is_err(),
"healthy Docs must not keep scheduling exchanges"
);
recovery.cancel.close();
recovery.join.await?;
let connections: Vec<_> = host_connections
.lock()
.unwrap()
.drain(..)
.chain(client_connections.lock().unwrap().drain(..))
.collect();
assert!(
!connections.is_empty(),
"recovery must restore the real gossip edge"
);
for connection in connections {
connection.close(0u32.into(), b"after cancellation");
}
tokio::time::timeout(Duration::from_secs(5), async {
while let Some(event) = observed.next().await {
if matches!(event?, LiveEvent::NeighborDown(_)) {
return Ok::<_, anyhow::Error>(());
}
}
anyhow::bail!("Docs stream ended before cancelled neighbor loss")
})
.await??;
replica
.set_bytes(
writer,
Bytes::from_static(b"after-cancel"),
Bytes::from_static(b"must remain local"),
)
.await?;
tokio::time::sleep(Duration::from_secs(1)).await;
assert!(
doc.get_exact(writer, b"after-cancel", false)
.await?
.is_none(),
"cancelled owner must not re-admit sync"
);
assert!(host_connections.lock().unwrap().is_empty());
assert!(client_connections.lock().unwrap().is_empty());
client_router.shutdown().await?;
host_router.shutdown().await?;
Ok(())
}
}