bug(archive): ArchiveImporter spins forever on a truncated archive instead of erroring

[Dusk1e]

Summary

ArchiveImporter's Stream::poll_next (crates/xmtp_archive/src/importer.rs) only leaves
its read loop when the reader is exhausted and nothing is buffered:

if amount == 0 && this.decoded.is_empty() {
    break;
}

If the archive ends part-way through an element, amount is 0 on every subsequent read
but this.decoded still holds the partial element, so neither condition for progress is
ever met again:

• the element_len != 0 && this.decoded.len() >= element_len branch can't fire — the
  remaining bytes are short of the promised length
• the break can't fire — decoded isn't empty

The loop then calls read() on an exhausted reader forever. It never returns Pending, so
it isn't a stall the executor can see — it's a tight busy loop that pins a core and never
yields.

The same happens when fewer than 4 bytes remain: element_len stays 0, the length-prefix
branch never fires, and decoded is non-empty.

Impact

Importing a truncated or corrupt archive hangs instead of returning an error. This is
reachable from every binding's public device-sync surface:

• bindings/mobile/src/mls/device_sync/mod.rs (iOS/Android)
• bindings/node/src/device_sync.rs
• bindings/wasm/src/device_sync.rs
• crates/xmtp_mls/src/worker/device_sync/worker.rs:718

The worker path is the one that worries me most: it builds the importer over a
StreamReader wrapped around the HTTP response body for the sync payload. A download cut
short by a dropped connection produces exactly this input, so it doesn't take a malformed
file on disk — an interrupted network read is enough to wedge the device-sync worker.

On WASM the consequence is worse, since a busy loop that never yields blocks the only
thread.

Reproduction

Header, then a zstd payload whose 4-byte length prefix promises 1024 bytes while the stream
carries 8:

let mut payload = 1024u32.to_le_bytes().to_vec();
payload.extend_from_slice(&[0u8; 8]);

let mut encoder = ZstdEncoder::new(Vec::new());
encoder.write_all(&payload).await.unwrap();
encoder.close().await.unwrap();

let mut archive = BACKUP_VERSION.to_le_bytes().to_vec();
archive.extend_from_slice(&[0u8; NONCE_SIZE]);
archive.extend_from_slice(&encoder.into_inner());

let reader: AsyncReader = Box::pin(futures::io::Cursor::new(archive));
ArchiveImporter::load(reader, &[0u8; 32]).await // never returns

Run under a timeout, this is killed rather than completing:

running 1 test
EXIT_CODE=124

Suggested fix

Treat exhaustion of the reader as terminal, and distinguish a clean end of stream from a
partial element:

if amount == 0 {
    if element_len == 0 && this.decoded.is_empty() {
        break;
    }
    return Poll::Ready(Some(Err(std::io::Error::from(
        std::io::ErrorKind::UnexpectedEof,
    )
    .into())));
}

ArchiveError already has an IO(#[from] std::io::Error) variant, so no new error type is
needed. Well-formed archives are unaffected: they always reach the
decoded.len() >= element_len branch and return before the reader runs dry.

Opening a PR with this fix and a regression test.

---

While reading this function I also noticed that element_len is a local, so it's reset on
every poll_next call even though this.decoded persists. If the reader returns Pending
after the 4-byte length prefix has been drained, the length is lost and the next poll reads
4 payload bytes as a fresh prefix. I've left that alone here to keep the change reviewable
and because I haven't reproduced it end to end — happy to open it separately if you'd like.