Store::import_events loses retry deduplication after destination reopen

[heyoub]

Summary

Store::import_events is idempotent while the destination handle remains open, but the same import appends duplicate destination events after the destination is closed and reopened.

This breaks crash/restart safety for a resumable read-model follower: a process can commit an import batch, crash before persisting its caller cursor, reopen, retry the same source range, and duplicate the events.

Observed with batpak = 0.10.0. The same behavior is present in the current local checkout at f51bac5a.

Minimal reproducer

use batpak::coordinate::Coordinate;
use batpak::event::EventKind;
use batpak::store::{
    ImportOptions, ImportSelector, Store, StoreConfig,
};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let source_dir = tempfile::tempdir()?;
    let dest_dir = tempfile::tempdir()?;

    let source = Store::open(StoreConfig::new(source_dir.path()))?;
    let coord = Coordinate::new("entity:source", "scope:demo")?;
    let kind = EventKind::custom(0xF, 0x81);
    source.append(&coord, kind, &serde_json::json!({"n": 1}))?;
    source.append(&coord, kind, &serde_json::json!({"n": 2}))?;

    let options = ImportOptions::new("stable-source")?;

    let dest = Store::open(StoreConfig::new(dest_dir.path()))?;
    let first = dest.import_events(&source, &ImportSelector::all(), &options)?;
    assert_eq!(first.imported, 2);
    let before_reopen = dest.by_entity("entity:source").len();
    dest.close()?;

    let dest = Store::open(StoreConfig::new(dest_dir.path()))?;
    let retry = dest.import_events(&source, &ImportSelector::all(), &options)?;

    assert_eq!(retry.imported, 0);       // fails: reports newly imported
    assert_eq!(retry.deduplicated, 2);   // fails
    assert_eq!(
        dest.by_entity("entity:source").len(),
        before_reopen,
    );                                  // fails: duplicates appended

    source.close()?;
    dest.close()?;
    Ok(())
}

A same-handle repeat currently behaves correctly. The failure appears only after destination reopen.

Likely contract gap

import_key_already_present checks both the in-memory/durable idempotency index and index.get_by_id(key), but after reopen neither path recognizes the prior import keys. The signed import provenance visible through IndexEntry::receipt_extensions() during the original process is also not available as a public, reopen-stable recovery mechanism for downstream callers.

Impact

• crash after append / before caller-cursor persistence duplicates replica events
• retry/resume cannot rely on BatPak's documented deterministic import idempotency across process restart
• live-follow consumers need an additional caller-owned materialization ledger

Texo reproduced this with an adversarial stale-cursor test. Its temporary workaround atomically appends each raw imported page with a bounded caller-owned replica-ledger event, then rebuilds the represented-source set from that ledger on reopen. The workaround is isolated behind a compatibility module so it can be deleted when this is fixed upstream.

Requested acceptance criteria

1. Add a two-process or close/reopen regression test around import_events.
2. Repeating the same source_namespace + source_event_id range after destination reopen appends zero events.
3. The retry report returns imported == 0 and deduplicated == source_event_count.
4. Preserve raw payload bytes/content hashes and existing import provenance semantics.
5. Confirm behavior when the caller resumes from an older ImportSelector::after(sequence) cursor.