From 11b68bd662c3dd2179364ff6237268dad69038d8 Mon Sep 17 00:00:00 2001 From: Reiers Date: Sat, 11 Jul 2026 15:02:04 +0200 Subject: [PATCH] fix(#119): persist pending mpool messages across daemon restart #47's rebroadcast loop tracked pending signed messages in memory only. A daemon restart threw away the pending set, so any message the user pushed pre-restart: - had a nonce consumed on the sender's on-chain account view, - was gossiped once but may not have been mined, - was no longer tracked (no retry, no confirm poll, no OnFailed fire). If the message wasn't mined before the restart, it was silently dead. The next MpoolPush from the same account then hit an 'expected nonce X, got X+1' gap and stalled indefinitely. #51 covers 'chain state can be rebuilt from the network.' Signed pending messages cannot be rebuilt (the key signed them once, only the daemon that pushed them has the bytes), so they need durable persistence. ## What's new net/mpool/persist.go: append-only JSONL journal + in-memory index. - 'add' lines carry the raw signed bytes (base64) + metadata. - 'retry' lines bump the rebroadcast counter + anchored epoch. - 'tombstone' lines mark confirmed / max-retries / Forget removals. - On Open, replay all lines; later entries win. Automatic compaction when tombstones dominate (compactMinLines=8, compactRatio=2), and on corrupt-tail recovery. - Fsync on every write so a crash after Publish keeps the message. Pool integration (net/mpool/mpool.go): - Config.PersistPath opts in. When set, New() opens the journal + re- registers every live entry into the in-memory pending map. - Publish() fsyncs an 'add' line after gossipsub push. If the fsync fails, we surface the error to the caller (safe to retry: identical bytes = same CID = same nonce). - Forget() writes a tombstone. - Close() flushes and releases the journal handle. - New Stats.Restored + Stats.PersistPath surface observability. Reconcile integration (net/mpool/reconcile.go): - Confirmed branch writes a tombstone (so restart doesn't re-register a message that already landed). - Max-retries branch writes a tombstone (OnFailed fires exactly once even across a crash-and-restart: on load the entry is gone, no restore). - Rebroadcast writes a retry line with the bumped counter + anchor. pkg/daemon wiring: - Config.MpoolPersistDisabled (zero value = enabled) toggles persistence. - Config.MpoolPersistPath overrides the default path. - start.go derives //mpool/pending.jsonl by default and passes it to mpool.New. - New Daemon.MpoolStats() accessor exposes counters for dashboard + tests. ## Guarantees preserved - Rebroadcast IS byte-identical (same nonce, same CID). We store the ORIGINAL raw bytes and never re-sign or bump fee. RBF stays out of scope. - Never touches private keys, JWT, or tokens. The journal lives in /mpool/, chain-side of the secrets boundary. - 'lantern reset --chain-state' does NOT wipe pending.jsonl (user's pending sends are user state; the reset allow-list only knows about 'headerstore' + 'bootstrap-anchor.json'). - Backwards-compatible: PersistPath='' preserves pre-#119 memory-only behaviour. All #47 tests still pass unchanged. - Nonce coordination is transparent: MpoolGetNonce already reads Pending() (rpc/handlers/chain_api.go), and the restored set is loaded before New() returns, so the very first MpoolGetNonce after restart sees the pre-outage pending nonces. ## Tests - 10 unit tests in net/mpool/persist_test.go: add/reload, remove-then-reload, update-on-rebroadcast reload, no-op on missing cid, tombstone-churn compaction, corrupt-tail recovery, journal-line shape, empty-path rejection, parent-dir mkdir. - 5 pool-level restart integration tests in net/mpool/persist_pool_test.go: restart-preserves-pending, confirmed-drops-across-restart, forget-drops-across-restart, retry-count-survives-restart, empty-path-is-memory-only. - All existing #47 reconcile tests + mpool tests remain green. - Full 'go test -count=1 -short ./...' green across the tree. ## Second half of the peace-of-mind contract Companion to #118 (auto-refresh stale anchor across restart). Together: a Lantern user can stop and restart the daemon at any time, after any duration, and the node catches up on its own AND their pending sends still land, without them needing to know a single reset command exists. Fixes #119. --- net/mpool/mpool.go | 111 +++++++- net/mpool/persist.go | 454 +++++++++++++++++++++++++++++++++ net/mpool/persist_pool_test.go | 278 ++++++++++++++++++++ net/mpool/persist_test.go | 257 +++++++++++++++++++ net/mpool/reconcile.go | 32 +++ pkg/daemon/daemon.go | 27 ++ pkg/daemon/start.go | 16 +- 7 files changed, 1171 insertions(+), 4 deletions(-) create mode 100644 net/mpool/persist.go create mode 100644 net/mpool/persist_pool_test.go create mode 100644 net/mpool/persist_test.go diff --git a/net/mpool/mpool.go b/net/mpool/mpool.go index f9e830a..15a568a 100644 --- a/net/mpool/mpool.go +++ b/net/mpool/mpool.go @@ -18,6 +18,7 @@ import ( "errors" "fmt" "sync" + "time" "github.com/ipfs/go-cid" logging "github.com/ipfs/go-log/v2" @@ -59,6 +60,28 @@ type Config struct { // (max retries exhausted). The message is also moved to a failed set // observable via Stats.Failed. Never silently stuck. OnFailed func(*ltypes.SignedMessage, string) + + // --- #119: durable pending-message store --- + // + // PersistPath is the on-disk JSONL journal Lantern uses to keep the + // pending set alive across daemon restart. When empty (the historical + // default), the pool is memory-only and behaves like pre-#119. + // + // When set: + // - New() opens the file, replays the journal, and re-registers every + // live entry into the in-memory pending set. Nonce derivation + // (MpoolGetNonce) reads Pending() so a restart transparently keeps + // each account's next nonce correct. + // - Publish() fsyncs an "add" line before returning success. + // - Reconcile confirm/fail branches fsync a "tombstone" line. + // - Rebroadcast fsyncs a "retry" line with the bumped counter. + // + // The file lives at //mpool/pending.jsonl. It is chain- + // side of the secrets boundary (never in keystore/, secrets/, or + // backups/), but `lantern reset --chain-state` deliberately does NOT + // wipe it: user-signed pending messages are user state, not rebuildable + // chain state. + PersistPath string } // pendingMsg tracks a locally published message for the confirm/retry loop. @@ -85,6 +108,12 @@ type Pool struct { rebroad uint64 // total rebroadcasts (#47) confirmd uint64 // total confirmed-on-chain (#47) failed uint64 // total given up (#47) + + // #119: durable pending-message journal. nil when PersistPath is empty. + persist *persistStore + // restored counts how many entries were re-registered from disk at + // startup (observable via Stats for smoke tests + dashboard). + restored uint64 } // New starts a Pool: joins the topic, subscribes, and (if OnMessage is set) @@ -115,14 +144,58 @@ func New(ctx context.Context, ps *pubsub.PubSub, cfg Config) (*Pool, error) { cfg: cfg, pending: make(map[cid.Cid]*pendingMsg), } + + // #119: if PersistPath is set, open the journal + re-register any + // live entries. Failure to open the store is fatal (we'd otherwise + // silently regress to memory-only behaviour and lose the peace-of-mind + // contract on the very restart that motivated #119). + if cfg.PersistPath != "" { + store, perr := openPersistStore(cfg.PersistPath) + if perr != nil { + _ = sub.Cancel + _ = t.Close() + return nil, fmt.Errorf("open pending journal: %w", perr) + } + p.persist = store + // Re-register live entries. Counter resets on restart: the + // resurrection itself is not a rebroadcast, only actual re-gossip + // events on the next Reconcile bump the counter. + for _, e := range store.All() { + sm, derr := e.SignedMessage() + if derr != nil { + log.Warnw("mpool persist: dropping undecodable entry", "cid", e.CID, "err", derr) + _ = store.Remove(e.CID) + continue + } + p.pending[e.CID] = &pendingMsg{ + sm: sm, + raw: e.Raw, + publishedAt: e.PublishedAt, + retries: e.Retries, + lastActivity: e.PublishedAt, + } + p.restored++ + } + if p.restored > 0 { + log.Infow("mpool persist: restored pending entries across restart", "count", p.restored, "path", cfg.PersistPath) + } + } + go p.readLoop(ctx) return p, nil } -// Close stops the subscription and topic. +// Close stops the subscription and topic, and closes the persist journal +// (when open). func (p *Pool) Close() error { p.sub.Cancel() - return p.topic.Close() + topicErr := p.topic.Close() + if p.persist != nil { + if perr := p.persist.Close(); perr != nil && topicErr == nil { + return perr + } + } + return topicErr } func (p *Pool) readLoop(ctx context.Context) { @@ -171,6 +244,18 @@ func (p *Pool) Publish(ctx context.Context, sm *ltypes.SignedMessage) (cid.Cid, p.mu.Lock() p.pending[mcid] = &pendingMsg{sm: sm, raw: raw} p.mu.Unlock() + // #119: persist even dry-run entries so tests exercise the same + // journal path production uses. A production caller with DryRun=true + // AND PersistPath set is unusual but consistent. + if p.persist != nil { + if perr := p.persist.Add(&PersistEntry{ + CID: mcid, + Raw: raw, + FirstSeenWall: time.Now().UTC(), + }); perr != nil { + log.Warnw("mpool persist: add failed on dry-run publish", "cid", mcid, "err", perr) + } + } log.Infof("mpool DRY-RUN: would publish %s (%d bytes) to %s", mcid, len(raw), p.cfg.Topic) return mcid, ErrDryRun } @@ -182,6 +267,19 @@ func (p *Pool) Publish(ctx context.Context, sm *ltypes.SignedMessage) (cid.Cid, p.pending[mcid] = &pendingMsg{sm: sm, raw: raw} p.publishd++ p.mu.Unlock() + // #119: fsync the entry to the journal AFTER the gossipsub push. If + // the fsync fails we surface as an error even though the message is + // already on the wire; the caller can decide whether to retry (which + // is idempotent: identical bytes, same nonce, same CID). + if p.persist != nil { + if perr := p.persist.Add(&PersistEntry{ + CID: mcid, + Raw: raw, + FirstSeenWall: time.Now().UTC(), + }); perr != nil { + return mcid, fmt.Errorf("persist pending message (gossipsub push already sent, safe to retry): %w", perr) + } + } return mcid, nil } @@ -203,6 +301,11 @@ func (p *Pool) Forget(c cid.Cid) { p.mu.Lock() delete(p.pending, c) p.mu.Unlock() + if p.persist != nil { + if err := p.persist.Remove(c); err != nil { + log.Warnw("mpool persist: remove on Forget failed", "cid", c, "err", err) + } + } } // Stats reports observable counters. @@ -215,6 +318,8 @@ type Stats struct { Failed uint64 // #47 PendingCnt int Topic string + Restored uint64 // #119 — entries re-registered from persist journal on startup + PersistPath string // #119 — empty when persistence is disabled } // Stats returns activity counters. @@ -230,6 +335,8 @@ func (p *Pool) Stats() Stats { Failed: p.failed, PendingCnt: len(p.pending), Topic: p.cfg.Topic, + Restored: p.restored, + PersistPath: p.cfg.PersistPath, } } diff --git a/net/mpool/persist.go b/net/mpool/persist.go new file mode 100644 index 0000000..c27860d --- /dev/null +++ b/net/mpool/persist.go @@ -0,0 +1,454 @@ +package mpool + +// #119: durable pending-message store. +// +// The problem: #47's rebroadcast loop tracks pending signed messages in +// memory. On daemon restart the pending set is thrown away. Messages the +// user pushed pre-restart: +// - already consumed a nonce on the sender's on-chain account view, +// - were gossiped once but may not have been mined, +// - are no longer tracked, no retry, no confirm poll, no OnFailed fire. +// +// If the message wasn't mined before the restart, it's silently dead. The +// user's next MpoolPush from the same account then hits an "expected +// nonce X, got X+1" gap and stalls indefinitely. +// +// #51 covers "chain state can be rebuilt from the network." Signed +// pending messages cannot be rebuilt (the key signed them once, only the +// daemon that pushed them has the bytes), so they live beside chain +// state but with the same durability property as user state. +// +// Design: append-only JSONL at //mpool/pending.jsonl. +// - Publish appends an `add` line with the raw signed bytes. +// - Confirm / max-retries-fail / Forget append a `tombstone` line. +// - Rebroadcast appends a `retry` line updating the counter + anchor. +// - On Open, replay all lines; later entries win. Compact when the +// tombstone-to-live ratio grows large. +// - Rebroadcast IS byte-identical (same nonce, same CID), so we never +// re-sign; we just store the ORIGINAL raw bytes. +// +// Never touches: +// - private keys (they live in /secrets/keystore, this file lives +// in /mpool/, chain-side of the secrets boundary), +// - JWT / tokens. +// +// `lantern reset --chain-state` must NOT wipe pending.jsonl: user's +// pending sends are user state, not rebuildable chain state. The reset +// allow-list only knows about `headerstore` + `bootstrap-anchor.json`, +// so this file is already safe from that path. + +import ( + "bufio" + "bytes" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sync" + "time" + + "github.com/ipfs/go-cid" + + ltypes "github.com/Reiers/lantern/chain/types" +) + +// PersistEntry is one live pending message in the durable store. Retries +// and PublishedAt are updated in place on rebroadcast; the raw bytes and +// FirstSeenWall are captured once at publish and never mutated. +type PersistEntry struct { + CID cid.Cid + Raw []byte + PublishedAt int64 + Retries int + FirstSeenWall time.Time +} + +// SignedMessage decodes the raw bytes back into a live SignedMessage. +// Returns an error if the bytes are corrupt (shouldn't happen for +// entries we wrote ourselves, but be defensive on load). +func (e *PersistEntry) SignedMessage() (*ltypes.SignedMessage, error) { + if len(e.Raw) == 0 { + return nil, errors.New("persist entry: empty raw bytes") + } + var sm ltypes.SignedMessage + if err := sm.UnmarshalCBOR(bytes.NewReader(e.Raw)); err != nil { + return nil, fmt.Errorf("persist entry: decode signed message: %w", err) + } + return &sm, nil +} + +// persistLine is the on-disk shape of one journal record. +type persistLine struct { + // Op is "add" | "retry" | "tombstone". + Op string `json:"op"` + // CID identifies the entry. + CID string `json:"cid"` + // RawB64 is base64(raw signed bytes); set only on "add". + RawB64 string `json:"raw,omitempty"` + // PublishedAt is the anchored epoch (set on add + retry). + PublishedAt int64 `json:"publishedAt,omitempty"` + // Retries is the current rebroadcast counter (set on retry). + Retries int `json:"retries,omitempty"` + // FirstSeenWall is the wall-clock time we first saw this cid. + FirstSeenWall time.Time `json:"firstSeenWall,omitzero"` +} + +// persistStore is the append-only journal + in-memory index. +type persistStore struct { + path string + + mu sync.Mutex + f *os.File + entries map[cid.Cid]*PersistEntry + // records counts total journal lines written (adds + retries + + // tombstones + carried-over on Open). When the live-entry ratio drops + // below compactRatio and there are > compactMinLines total, we compact. + records int +} + +// compactRatio is the live-to-total ratio above which the store rewrites +// the journal in a single compact pass. Small file, cheap rewrite; keeps +// pathological append-only growth bounded when a busy sender churns. +// +// compactMinLines is the small-file floor: below it, compaction skips +// (the file is already tiny; rewriting saves nothing). 8 lines is roughly +// one full turnaround of a small pending set. +const ( + compactRatio = 2 // compact when records > liveEntries * compactRatio + compactMinLines = 8 +) + +// openPersistStore opens (or creates) the durable pending journal at +// path. The parent directory must already exist. Returns a store with +// the current live set already loaded into memory. +// +// If the journal is corrupt at some tail line, prior valid lines are +// still loaded; the corrupt tail is dropped and the file is compacted +// on open so subsequent writes don't inherit the damage. +func openPersistStore(path string) (*persistStore, error) { + if path == "" { + return nil, errors.New("mpool persist: empty path") + } + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return nil, fmt.Errorf("mpool persist: mkdir: %w", err) + } + s := &persistStore{ + path: path, + entries: make(map[cid.Cid]*PersistEntry), + } + if err := s.loadAndOpen(); err != nil { + return nil, err + } + return s, nil +} + +// loadAndOpen replays the existing journal into memory, then keeps the +// file open in append mode for subsequent writes. Compacts on open when +// the journal is dominated by tombstones (or corrupt at the tail). +func (s *persistStore) loadAndOpen() error { + f, err := os.OpenFile(s.path, os.O_RDONLY|os.O_CREATE, 0o600) + if err != nil { + return fmt.Errorf("mpool persist: open read: %w", err) + } + + corruptTail := false + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024) // signed messages fit well under 4 MiB + lines := 0 + for sc.Scan() { + line := sc.Bytes() + if len(line) == 0 { + continue + } + lines++ + var rec persistLine + if err := json.Unmarshal(line, &rec); err != nil { + // Corrupt tail (partial write, disk corruption). Drop it + + // mark for compaction so we don't inherit the damage. + corruptTail = true + log.Warnw("mpool persist: dropping corrupt journal line", "path", s.path, "err", err) + continue + } + if err := s.applyRecord(rec); err != nil { + corruptTail = true + log.Warnw("mpool persist: dropping invalid journal record", "path", s.path, "err", err) + } + } + if err := sc.Err(); err != nil { + // Scanner error mid-stream (usually EOF at partial line, which + // bufio surfaces as ErrUnexpectedEOF via the scanner). Treat as + // corrupt tail: keep whatever we managed to load, compact. + if !errors.Is(err, io.ErrUnexpectedEOF) { + log.Warnw("mpool persist: journal scanner error, keeping partial load", "path", s.path, "err", err) + } + corruptTail = true + } + _ = f.Close() + s.records = lines + + // Decide whether to compact. Rewrite in place when corrupt tail was + // observed or when tombstones dominate the file. + shouldCompact := corruptTail || (lines >= compactMinLines && lines > len(s.entries)*compactRatio) + if shouldCompact { + if err := s.compactLocked(); err != nil { + return fmt.Errorf("mpool persist: compact on open: %w", err) + } + } + + // Reopen for append (create if we just compacted / cold-open). + appendF, err := os.OpenFile(s.path, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600) + if err != nil { + return fmt.Errorf("mpool persist: open append: %w", err) + } + s.f = appendF + return nil +} + +// applyRecord folds one journal line into the in-memory index. +func (s *persistStore) applyRecord(rec persistLine) error { + c, err := cid.Parse(rec.CID) + if err != nil { + return fmt.Errorf("parse cid %q: %w", rec.CID, err) + } + switch rec.Op { + case "add": + raw, derr := base64.StdEncoding.DecodeString(rec.RawB64) + if derr != nil { + return fmt.Errorf("decode raw: %w", derr) + } + s.entries[c] = &PersistEntry{ + CID: c, + Raw: raw, + PublishedAt: rec.PublishedAt, + Retries: rec.Retries, + FirstSeenWall: rec.FirstSeenWall, + } + case "retry": + if e, ok := s.entries[c]; ok { + e.PublishedAt = rec.PublishedAt + e.Retries = rec.Retries + } + // A retry for a cid we never saw is discarded (tombstone was + // applied earlier, or add line was corrupt). Benign. + case "tombstone": + delete(s.entries, c) + default: + return fmt.Errorf("unknown op %q", rec.Op) + } + return nil +} + +// compactLocked rewrites the journal to contain exactly one "add" line +// per live entry. Must be called with s.mu held OR before s.f is set. +func (s *persistStore) compactLocked() error { + tmpPath := s.path + ".compact" + tmp, err := os.OpenFile(tmpPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) + if err != nil { + return fmt.Errorf("open compact tmp: %w", err) + } + w := bufio.NewWriter(tmp) + written := 0 + for _, e := range s.entries { + rec := persistLine{ + Op: "add", + CID: e.CID.String(), + RawB64: base64.StdEncoding.EncodeToString(e.Raw), + PublishedAt: e.PublishedAt, + Retries: e.Retries, + FirstSeenWall: e.FirstSeenWall, + } + b, merr := json.Marshal(&rec) + if merr != nil { + _ = tmp.Close() + _ = os.Remove(tmpPath) + return fmt.Errorf("marshal compact rec: %w", merr) + } + if _, werr := w.Write(append(b, '\n')); werr != nil { + _ = tmp.Close() + _ = os.Remove(tmpPath) + return fmt.Errorf("write compact rec: %w", werr) + } + written++ + } + if err := w.Flush(); err != nil { + _ = tmp.Close() + _ = os.Remove(tmpPath) + return fmt.Errorf("flush compact: %w", err) + } + if err := tmp.Sync(); err != nil { + _ = tmp.Close() + _ = os.Remove(tmpPath) + return fmt.Errorf("fsync compact: %w", err) + } + if err := tmp.Close(); err != nil { + _ = os.Remove(tmpPath) + return fmt.Errorf("close compact tmp: %w", err) + } + // Close the current append handle before rename (Windows-safe; + // darwin/linux tolerant either way). + if s.f != nil { + _ = s.f.Close() + s.f = nil + } + if err := os.Rename(tmpPath, s.path); err != nil { + return fmt.Errorf("rename compact: %w", err) + } + s.records = written + // Caller (loadAndOpen or writeLocked) reopens the append handle. + return nil +} + +// writeLineLocked appends one line to the journal and fsyncs. Must be +// called with s.mu held. +func (s *persistStore) writeLineLocked(rec persistLine) error { + if s.f == nil { + return errors.New("mpool persist: store closed") + } + b, err := json.Marshal(&rec) + if err != nil { + return fmt.Errorf("marshal record: %w", err) + } + b = append(b, '\n') + if _, err := s.f.Write(b); err != nil { + return fmt.Errorf("append record: %w", err) + } + if err := s.f.Sync(); err != nil { + return fmt.Errorf("fsync record: %w", err) + } + s.records++ + return nil +} + +// maybeCompactLocked triggers a compact when tombstones dominate the +// file. Must be called with s.mu held. +func (s *persistStore) maybeCompactLocked() { + if s.records < compactMinLines { + return + } + live := len(s.entries) + if live == 0 { + // Everything's tombstoned; rewrite to an empty file. + if err := s.compactLocked(); err != nil { + log.Warnw("mpool persist: compact-empty failed", "path", s.path, "err", err) + return + } + f, err := os.OpenFile(s.path, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600) + if err != nil { + log.Warnw("mpool persist: reopen after compact-empty failed", "path", s.path, "err", err) + return + } + s.f = f + return + } + if s.records > live*compactRatio { + if err := s.compactLocked(); err != nil { + log.Warnw("mpool persist: compact failed", "path", s.path, "err", err) + return + } + f, err := os.OpenFile(s.path, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600) + if err != nil { + log.Warnw("mpool persist: reopen after compact failed", "path", s.path, "err", err) + return + } + s.f = f + } +} + +// Add records a new pending message. On success the fsync has landed. +func (s *persistStore) Add(e *PersistEntry) error { + if e == nil { + return errors.New("nil entry") + } + s.mu.Lock() + defer s.mu.Unlock() + rec := persistLine{ + Op: "add", + CID: e.CID.String(), + RawB64: base64.StdEncoding.EncodeToString(e.Raw), + PublishedAt: e.PublishedAt, + Retries: e.Retries, + FirstSeenWall: e.FirstSeenWall, + } + if err := s.writeLineLocked(rec); err != nil { + return err + } + // Copy so caller mutations don't leak into the in-memory index. + cpy := *e + s.entries[e.CID] = &cpy + return nil +} + +// Remove tombstones an entry. Idempotent on a missing cid (writes a +// tombstone line so a subsequent compact drops the empty entry cleanly). +func (s *persistStore) Remove(c cid.Cid) error { + s.mu.Lock() + defer s.mu.Unlock() + // Only journal a tombstone when the entry is actually present, so + // we don't grow the file with no-op tombstones on repeated Forget. + if _, ok := s.entries[c]; !ok { + return nil + } + rec := persistLine{ + Op: "tombstone", + CID: c.String(), + } + if err := s.writeLineLocked(rec); err != nil { + return err + } + delete(s.entries, c) + s.maybeCompactLocked() + return nil +} + +// UpdateOnRebroadcast bumps the retries counter + anchored epoch for an +// existing entry. No-op (with no journal line) when the cid isn't +// present (already tombstoned by a concurrent Reconcile pass). +func (s *persistStore) UpdateOnRebroadcast(c cid.Cid, retries int, publishedAt int64) error { + s.mu.Lock() + defer s.mu.Unlock() + e, ok := s.entries[c] + if !ok { + return nil + } + rec := persistLine{ + Op: "retry", + CID: c.String(), + Retries: retries, + PublishedAt: publishedAt, + } + if err := s.writeLineLocked(rec); err != nil { + return err + } + e.Retries = retries + e.PublishedAt = publishedAt + s.maybeCompactLocked() + return nil +} + +// All returns a stable snapshot of live entries at call time. +func (s *persistStore) All() []*PersistEntry { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]*PersistEntry, 0, len(s.entries)) + for _, e := range s.entries { + cpy := *e + out = append(out, &cpy) + } + return out +} + +// Close flushes and closes the underlying file. Safe to call multiple +// times. +func (s *persistStore) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.f == nil { + return nil + } + err := s.f.Close() + s.f = nil + return err +} diff --git a/net/mpool/persist_pool_test.go b/net/mpool/persist_pool_test.go new file mode 100644 index 0000000..5869a22 --- /dev/null +++ b/net/mpool/persist_pool_test.go @@ -0,0 +1,278 @@ +// Pool-level restart tests for #119. +// +// These stand up a real Pool (with a real libp2p host + pubsub, DryRun +// mode so nothing hits the network) against a temp-dir persist path, +// publish some messages, close the pool, reopen a NEW Pool at the same +// path, and assert the pending set was restored across the "restart". +package mpool_test + +import ( + "context" + "errors" + "path/filepath" + "testing" + "time" + + "github.com/filecoin-project/go-address" + "github.com/filecoin-project/go-state-types/big" + gscrypto "github.com/filecoin-project/go-state-types/crypto" + "github.com/ipfs/go-cid" + "github.com/stretchr/testify/require" + + ltypes "github.com/Reiers/lantern/chain/types" + llibp2p "github.com/Reiers/lantern/net/libp2p" + "github.com/Reiers/lantern/net/mpool" +) + +// mkSignedNonce is like mkSigned in mpool_test.go but with a custom +// nonce so restart tests can push multiple distinct messages from the +// same account without collision. +func mkSignedNonce(t *testing.T, nonce uint64) *ltypes.SignedMessage { + t.Helper() + from, err := address.NewIDAddress(1000) + require.NoError(t, err) + to, err := address.NewIDAddress(1001) + require.NoError(t, err) + return <ypes.SignedMessage{ + Message: ltypes.Message{ + Version: 0, + From: from, + To: to, + Nonce: nonce, + Value: big.NewInt(1_000_000_000), + GasLimit: 10_000_000, + GasFeeCap: big.NewInt(100_000_000), + GasPremium: big.NewInt(100_000), + Method: 0, + }, + Signature: gscrypto.Signature{Type: gscrypto.SigTypeBLS, Data: []byte("fake-sig-bytes-96-long")}, + } +} + +// newDryPoolPersist stands up a fresh libp2p host + DryRun pool with a +// persist path pointing at `path`. Each call spins its own host so the +// "restart" case truly starts a new pool from scratch. +func newDryPoolPersist(t *testing.T, ctx context.Context, path string, cfg mpool.Config) *mpool.Pool { + t.Helper() + h, err := llibp2p.New(ctx, llibp2p.HostConfig{ListenAddrs: []string{"/ip4/127.0.0.1/tcp/0"}}) + require.NoError(t, err) + t.Cleanup(func() { h.Close() }) + cfg.DryRun = true + cfg.PersistPath = path + if cfg.Topic == "" { + cfg.Topic = "/fil/msgs/test-persist" + } + p, err := mpool.New(ctx, h.PubSub, cfg) + require.NoError(t, err) + t.Cleanup(func() { p.Close() }) + return p +} + +// TestPersist_RestartPreservesPending is the core #119 property test: +// publish 3 messages, close the pool, open a new Pool at the same path, +// assert Pending() returns all 3. +func TestPersist_RestartPreservesPending(t *testing.T) { + if testing.Short() { + t.Skip("skipping in short mode") + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + path := filepath.Join(t.TempDir(), "mpool", "pending.jsonl") + + // First "session": publish 3 messages. + { + p := newDryPoolPersist(t, ctx, path, mpool.Config{}) + for _, n := range []uint64{100, 101, 102} { + _, err := p.Publish(ctx, mkSignedNonce(t, n)) + require.True(t, errors.Is(err, mpool.ErrDryRun)) + } + require.Equal(t, 3, p.Stats().PendingCnt) + require.NoError(t, p.Close()) + } + + // Second "session": fresh Pool at the same persist path. All 3 + // must be restored into the pending set. + { + p := newDryPoolPersist(t, ctx, path, mpool.Config{}) + st := p.Stats() + require.Equal(t, 3, st.PendingCnt, "restart must restore pending set") + require.Equal(t, uint64(3), st.Restored, "Stats.Restored must reflect restore count") + require.Equal(t, path, st.PersistPath) + + pending := p.Pending() + require.Len(t, pending, 3) + nonces := map[uint64]bool{} + for _, sm := range pending { + nonces[sm.Message.Nonce] = true + } + require.True(t, nonces[100]) + require.True(t, nonces[101]) + require.True(t, nonces[102]) + } +} + +// TestPersist_ConfirmedDropsAcrossRestart: publish 2, confirm 1 via +// Reconcile, close, reopen. Only 1 must be restored. +func TestPersist_ConfirmedDropsAcrossRestart(t *testing.T) { + if testing.Short() { + t.Skip("skipping in short mode") + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + path := filepath.Join(t.TempDir(), "mpool", "pending.jsonl") + + var confirmedCID cid.Cid + { + p := newDryPoolPersist(t, ctx, path, mpool.Config{}) + c1, err := p.Publish(ctx, mkSignedNonce(t, 200)) + require.True(t, errors.Is(err, mpool.ErrDryRun)) + _, err = p.Publish(ctx, mkSignedNonce(t, 201)) + require.True(t, errors.Is(err, mpool.ErrDryRun)) + confirmedCID = c1 + + // Confirm the first message on chain via a targeted search. + p.Reconcile(ctx, 500, func(_ context.Context, mc cid.Cid) (mpool.SearchResult, error) { + if mc == confirmedCID { + return mpool.SearchFound, nil + } + return mpool.SearchUnknown, nil + }) + st := p.Stats() + require.Equal(t, 1, st.PendingCnt) + require.Equal(t, uint64(1), st.Confirmed) + require.NoError(t, p.Close()) + } + + // Restart: only the un-confirmed message must come back. + { + p := newDryPoolPersist(t, ctx, path, mpool.Config{}) + require.Equal(t, 1, p.Stats().PendingCnt) + require.Equal(t, uint64(1), p.Stats().Restored) + // The one that came back must NOT be the confirmed cid. + got := p.Pending() + require.Len(t, got, 1) + // Verify it's the surviving nonce (201). + require.Equal(t, uint64(201), got[0].Message.Nonce) + } +} + +// TestPersist_ForgetDropsAcrossRestart mirrors the confirm test but via +// the imperative Forget path (used by callers that confirm via a +// non-Reconcile mechanism). +func TestPersist_ForgetDropsAcrossRestart(t *testing.T) { + if testing.Short() { + t.Skip("skipping in short mode") + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + path := filepath.Join(t.TempDir(), "mpool", "pending.jsonl") + + { + p := newDryPoolPersist(t, ctx, path, mpool.Config{}) + c, err := p.Publish(ctx, mkSignedNonce(t, 300)) + require.True(t, errors.Is(err, mpool.ErrDryRun)) + p.Forget(c) + require.Equal(t, 0, p.Stats().PendingCnt) + require.NoError(t, p.Close()) + } + + { + p := newDryPoolPersist(t, ctx, path, mpool.Config{}) + require.Equal(t, 0, p.Stats().PendingCnt) + } +} + +// TestPersist_RetryCountSurvivesRestart: publish, run Reconcile enough +// times to rebroadcast, close, reopen; assert the retries counter was +// persisted. +func TestPersist_RetryCountSurvivesRestart(t *testing.T) { + if testing.Short() { + t.Skip("skipping in short mode") + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + path := filepath.Join(t.TempDir(), "mpool", "pending.jsonl") + + { + p := newDryPoolPersist(t, ctx, path, mpool.Config{ + ConfirmAfterEpochs: 3, + MaxRetries: 5, + }) + _, err := p.Publish(ctx, mkSignedNonce(t, 400)) + require.True(t, errors.Is(err, mpool.ErrDryRun)) + unknown := func(_ context.Context, _ cid.Cid) (mpool.SearchResult, error) { + return mpool.SearchUnknown, nil + } + p.Reconcile(ctx, 100, unknown) // anchors publishedAt=100 + p.Reconcile(ctx, 110, unknown) // retries=1 + p.Reconcile(ctx, 120, unknown) // retries=2 + require.Equal(t, uint64(2), p.Stats().Rebroadcasts) + require.NoError(t, p.Close()) + } + + // Fresh pool: the message is restored WITH its retry counter. + // One more past-window Reconcile should bring it to 3, then 4, 5, + // then fail at 6. + { + var failed bool + p := newDryPoolPersist(t, ctx, path, mpool.Config{ + ConfirmAfterEpochs: 3, + MaxRetries: 5, + OnFailed: func(*ltypes.SignedMessage, string) { failed = true }, + }) + require.Equal(t, 1, p.Stats().PendingCnt) + unknown := func(_ context.Context, _ cid.Cid) (mpool.SearchResult, error) { + return mpool.SearchUnknown, nil + } + // Anchor is now epoch 0 in memory (restored publishedAt=100 + // from journal, but Reconcile treats it as valid anchor). + p.Reconcile(ctx, 130, unknown) // retries=3 + p.Reconcile(ctx, 140, unknown) // retries=4 + p.Reconcile(ctx, 150, unknown) // retries=5 + require.Equal(t, 1, p.Stats().PendingCnt, "not yet failed at retries=5") + p.Reconcile(ctx, 160, unknown) // retries>=max -> failed + require.Equal(t, 0, p.Stats().PendingCnt) + require.Equal(t, uint64(1), p.Stats().Failed) + require.True(t, failed, "OnFailed must fire after retry counter crosses MaxRetries across restart") + } +} + +// TestPersist_EmptyPathIsMemoryOnly: PersistPath="" means #119 is off +// (backwards-compatible default). +func TestPersist_EmptyPathIsMemoryOnly(t *testing.T) { + if testing.Short() { + t.Skip("skipping in short mode") + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + // Two sequential pools with the same DryRun config; no path. + // Whatever the first publishes must NOT show up in the second. + h1, err := llibp2p.New(ctx, llibp2p.HostConfig{ListenAddrs: []string{"/ip4/127.0.0.1/tcp/0"}}) + require.NoError(t, err) + defer h1.Close() + p1, err := mpool.New(ctx, h1.PubSub, mpool.Config{ + Topic: "/fil/msgs/test-nopersist", + DryRun: true, + }) + require.NoError(t, err) + _, _ = p1.Publish(ctx, mkSignedNonce(t, 500)) + require.Equal(t, 1, p1.Stats().PendingCnt) + require.Empty(t, p1.Stats().PersistPath) + require.NoError(t, p1.Close()) + + h2, err := llibp2p.New(ctx, llibp2p.HostConfig{ListenAddrs: []string{"/ip4/127.0.0.1/tcp/0"}}) + require.NoError(t, err) + defer h2.Close() + p2, err := mpool.New(ctx, h2.PubSub, mpool.Config{ + Topic: "/fil/msgs/test-nopersist", + DryRun: true, + }) + require.NoError(t, err) + require.Equal(t, 0, p2.Stats().PendingCnt, "no persist path → nothing to restore") + require.NoError(t, p2.Close()) +} diff --git a/net/mpool/persist_test.go b/net/mpool/persist_test.go new file mode 100644 index 0000000..3d2509e --- /dev/null +++ b/net/mpool/persist_test.go @@ -0,0 +1,257 @@ +// Unit tests for the durable pending-message store (#119). +// +// These are pure filesystem tests — no libp2p host, no gossipsub. They +// exercise the JSONL append + tombstone + compact + reopen paths in +// isolation from the Pool integration. +package mpool + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/ipfs/go-cid" + mh "github.com/multiformats/go-multihash" + "github.com/stretchr/testify/require" +) + +// mkCID makes a deterministic CID from a small string, so tests are +// readable without wiring the CBOR-encoding path. +func mkCID(t *testing.T, s string) cid.Cid { + t.Helper() + h, err := mh.Sum([]byte(s), mh.SHA2_256, -1) + require.NoError(t, err) + return cid.NewCidV1(cid.Raw, h) +} + +// mkEntry returns a PersistEntry with valid-shape raw bytes. Real raw +// bytes come from ltypes.SignedMessage.Serialize in production; here we +// just need something the store can round-trip. +func mkEntry(t *testing.T, s string) *PersistEntry { + t.Helper() + return &PersistEntry{ + CID: mkCID(t, s), + Raw: []byte("raw-bytes-for-" + s), + PublishedAt: 0, + Retries: 0, + FirstSeenWall: time.Date(2026, 7, 11, 12, 0, 0, 0, time.UTC), + } +} + +func TestPersist_AddAndReload(t *testing.T) { + path := filepath.Join(t.TempDir(), "pending.jsonl") + s, err := openPersistStore(path) + require.NoError(t, err) + + require.NoError(t, s.Add(mkEntry(t, "one"))) + require.NoError(t, s.Add(mkEntry(t, "two"))) + require.NoError(t, s.Add(mkEntry(t, "three"))) + require.NoError(t, s.Close()) + + // Reopen: the three entries must come back. + s2, err := openPersistStore(path) + require.NoError(t, err) + defer s2.Close() + + all := s2.All() + require.Len(t, all, 3) + seen := map[string]bool{} + for _, e := range all { + seen[e.CID.String()] = true + require.Contains(t, string(e.Raw), "raw-bytes-for-") + } + require.True(t, seen[mkCID(t, "one").String()]) + require.True(t, seen[mkCID(t, "two").String()]) + require.True(t, seen[mkCID(t, "three").String()]) +} + +func TestPersist_RemoveThenReload(t *testing.T) { + path := filepath.Join(t.TempDir(), "pending.jsonl") + s, err := openPersistStore(path) + require.NoError(t, err) + + e1 := mkEntry(t, "a") + e2 := mkEntry(t, "b") + require.NoError(t, s.Add(e1)) + require.NoError(t, s.Add(e2)) + require.NoError(t, s.Remove(e1.CID)) + require.NoError(t, s.Close()) + + s2, err := openPersistStore(path) + require.NoError(t, err) + defer s2.Close() + + all := s2.All() + require.Len(t, all, 1) + require.Equal(t, e2.CID, all[0].CID) +} + +func TestPersist_UpdateOnRebroadcastReload(t *testing.T) { + path := filepath.Join(t.TempDir(), "pending.jsonl") + s, err := openPersistStore(path) + require.NoError(t, err) + + e := mkEntry(t, "retryable") + require.NoError(t, s.Add(e)) + require.NoError(t, s.UpdateOnRebroadcast(e.CID, 3, 4242)) + require.NoError(t, s.Close()) + + s2, err := openPersistStore(path) + require.NoError(t, err) + defer s2.Close() + + all := s2.All() + require.Len(t, all, 1) + require.Equal(t, 3, all[0].Retries) + require.Equal(t, int64(4242), all[0].PublishedAt) +} + +// TestPersist_RemoveOnMissingCID is a no-op — nothing to journal. +func TestPersist_RemoveOnMissingCID(t *testing.T) { + path := filepath.Join(t.TempDir(), "pending.jsonl") + s, err := openPersistStore(path) + require.NoError(t, err) + defer s.Close() + + // Remove without any Add: no error, journal stays empty. + require.NoError(t, s.Remove(mkCID(t, "ghost"))) + + info, err := os.Stat(path) + require.NoError(t, err) + require.Equal(t, int64(0), info.Size(), "no-op Remove must not journal a line") +} + +// TestPersist_UpdateOnRebroadcastMissingCID is a no-op. +func TestPersist_UpdateOnRebroadcastMissingCID(t *testing.T) { + path := filepath.Join(t.TempDir(), "pending.jsonl") + s, err := openPersistStore(path) + require.NoError(t, err) + defer s.Close() + + // Retry-update on an unknown cid must not journal a line. + require.NoError(t, s.UpdateOnRebroadcast(mkCID(t, "ghost"), 5, 100)) + info, err := os.Stat(path) + require.NoError(t, err) + require.Equal(t, int64(0), info.Size()) +} + +// TestPersist_CompactAfterTombstoneChurn: churn > compactMinLines total +// records with tombstone-dominant ratio, then reopen and verify the file +// was compacted (line count matches the live set). +func TestPersist_CompactAfterTombstoneChurn(t *testing.T) { + path := filepath.Join(t.TempDir(), "pending.jsonl") + s, err := openPersistStore(path) + require.NoError(t, err) + + // 20 adds + 15 removes: 35 lines total, 5 live. records/live > 2. + entries := make([]*PersistEntry, 20) + for i := range entries { + entries[i] = mkEntry(t, "entry-"+string(rune('a'+i))) + require.NoError(t, s.Add(entries[i])) + } + for _, e := range entries[:15] { + require.NoError(t, s.Remove(e.CID)) + } + // Compaction should have fired inside Remove. Verify by counting + // lines in the file. + require.NoError(t, s.Close()) + + raw, err := os.ReadFile(path) + require.NoError(t, err) + lines := 0 + for _, ln := range strings.Split(string(raw), "\n") { + if strings.TrimSpace(ln) != "" { + lines++ + } + } + require.LessOrEqual(t, lines, 6, "after compact, journal must contain roughly the live set (5), got %d lines", lines) + + // Reload: exactly 5 live entries. + s2, err := openPersistStore(path) + require.NoError(t, err) + defer s2.Close() + require.Len(t, s2.All(), 5) +} + +// TestPersist_CorruptTailIsRecovered writes an intentionally malformed +// last line, then verifies open logs the drop, keeps the earlier live +// entries, and compacts the file so the corruption doesn't survive. +func TestPersist_CorruptTailIsRecovered(t *testing.T) { + path := filepath.Join(t.TempDir(), "pending.jsonl") + s, err := openPersistStore(path) + require.NoError(t, err) + + e := mkEntry(t, "keeper") + require.NoError(t, s.Add(e)) + require.NoError(t, s.Close()) + + // Append a corrupt tail line to simulate a torn write. + f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0o600) + require.NoError(t, err) + _, err = f.WriteString("{ this is not valid json\n") + require.NoError(t, err) + require.NoError(t, f.Close()) + + // Reopen: the keeper survives, the corrupt line is dropped. + s2, err := openPersistStore(path) + require.NoError(t, err) + defer s2.Close() + require.Len(t, s2.All(), 1) + require.Equal(t, e.CID, s2.All()[0].CID) + + // After the corrupt-tail-triggered compaction, the file must not + // contain the malformed line anymore. + raw, rerr := os.ReadFile(path) + require.NoError(t, rerr) + require.NotContains(t, string(raw), "not valid json") +} + +// TestPersist_JournalLineShape sanity-checks the on-disk format, so a +// future refactor that changes field names surfaces here instead of at +// upgrade time on user machines. +func TestPersist_JournalLineShape(t *testing.T) { + path := filepath.Join(t.TempDir(), "pending.jsonl") + s, err := openPersistStore(path) + require.NoError(t, err) + + e := mkEntry(t, "shape") + require.NoError(t, s.Add(e)) + require.NoError(t, s.Close()) + + raw, err := os.ReadFile(path) + require.NoError(t, err) + line := strings.TrimSpace(strings.Split(string(raw), "\n")[0]) + + var rec persistLine + require.NoError(t, json.Unmarshal([]byte(line), &rec)) + require.Equal(t, "add", rec.Op) + require.Equal(t, e.CID.String(), rec.CID) + require.NotEmpty(t, rec.RawB64) +} + +// TestPersist_EmptyPathRejected: empty path is a programmer error. +func TestPersist_EmptyPathRejected(t *testing.T) { + _, err := openPersistStore("") + require.Error(t, err) + require.Contains(t, err.Error(), "empty path") +} + +// TestPersist_MkdirParent verifies the store creates its parent +// directory (matches the //mpool/ convention where the +// mpool dir may not exist on first Pool.New). +func TestPersist_MkdirParent(t *testing.T) { + root := t.TempDir() + path := filepath.Join(root, "does-not-exist-yet", "mpool", "pending.jsonl") + + s, err := openPersistStore(path) + require.NoError(t, err) + defer s.Close() + require.NoError(t, s.Add(mkEntry(t, "in-a-fresh-dir"))) + + info, err := os.Stat(path) + require.NoError(t, err) + require.Greater(t, info.Size(), int64(0)) +} diff --git a/net/mpool/reconcile.go b/net/mpool/reconcile.go index 28d229b..59d3ee9 100644 --- a/net/mpool/reconcile.go +++ b/net/mpool/reconcile.go @@ -86,6 +86,13 @@ func (p *Pool) Reconcile(ctx context.Context, headEpoch int64, search SearchFunc p.confirmd++ } p.mu.Unlock() + // #119: drop from durable journal so a subsequent restart + // doesn't re-register a message that already landed. + if p.persist != nil { + if rerr := p.persist.Remove(it.cid); rerr != nil { + log.Warnw("mpool persist: remove on confirm failed", "cid", it.cid, "err", rerr) + } + } log.Debugw("mpool: message confirmed on chain", "cid", it.cid) continue } @@ -106,6 +113,15 @@ func (p *Pool) Reconcile(ctx context.Context, headEpoch int64, search SearchFunc p.failed++ } p.mu.Unlock() + // #119: drop from durable journal on give-up. OnFailed fires + // exactly once (before the tombstone hits disk, but the tombstone + // is idempotent so a crash-and-restart never re-fires OnFailed + // for the same cid: on load the entry is gone → no restore). + if ok && p.persist != nil { + if rerr := p.persist.Remove(it.cid); rerr != nil { + log.Warnw("mpool persist: remove on give-up failed", "cid", it.cid, "err", rerr) + } + } if ok { log.Warnw("mpool: message gave up (max retries) — surfacing as failed", "cid", it.cid, "retries", pm.retries, "nonce", pm.sm.Message.Nonce) @@ -125,14 +141,30 @@ func (p *Pool) Reconcile(ctx context.Context, headEpoch int64, search SearchFunc continue } } + var ( + newRetries int + anchor int64 + stillLive bool + ) p.mu.Lock() if pm, ok := p.pending[it.cid]; ok { pm.retries++ pm.lastActivity = headEpoch pm.publishedAt = it.pm.publishedAt // keep original anchor; window resets via lastActivity if desired p.rebroad++ + newRetries = pm.retries + anchor = pm.publishedAt + stillLive = true } p.mu.Unlock() + // #119: persist the bumped retry counter so a restart doesn't + // silently reset progress toward MaxRetries. Idempotent no-op if + // the entry was tombstoned concurrently. + if stillLive && p.persist != nil { + if uerr := p.persist.UpdateOnRebroadcast(it.cid, newRetries, anchor); uerr != nil { + log.Warnw("mpool persist: update on rebroadcast failed", "cid", it.cid, "err", uerr) + } + } log.Debugw("mpool: rebroadcast pending message (identical bytes)", "cid", it.cid, "retries", it.pm.retries+1) } } diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 174d90f..5f58c78 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -250,6 +250,18 @@ type Config struct { // (2 retries / 8s total). Set FEVMFetchRetries < 0 to disable. FEVMFetchRetries int FEVMFetchTimeout time.Duration + + // MpoolPersistDisabled turns OFF the durable pending-message store + // (#119). Zero value = enabled: user pushed messages survive daemon + // restart, and the sender's next nonce stays consistent because + // MpoolGetNonce reads the restored pending set. Set to true only for + // stateless embedded uses (CI, ephemeral tests) that don't want the + // per-network mpool directory. + MpoolPersistDisabled bool + + // MpoolPersistPath overrides the default persist journal path + // (//mpool/pending.jsonl). Empty uses the default. + MpoolPersistPath string } // applyDefaults populates zero-value fields with the same defaults @@ -484,6 +496,21 @@ func (d *Daemon) LocalEthCallStats() (handlers.LocalEthCallStatsView, bool) { return ch.LocalEthCallStatsView(), true } +// MpoolStats returns a snapshot of gossipsub-mempool counters (including +// #119 Restored + PersistPath) and true when the mempool is wired. +// Returns (zero, false) when the pool wasn't wired (e.g. libp2p disabled +// or mpool init failure). Useful for verifying restart-persistence +// behavior after a soft restart. +func (d *Daemon) MpoolStats() (mpool.Stats, bool) { + d.mu.Lock() + mp := d.mpool + d.mu.Unlock() + if mp == nil { + return mpool.Stats{}, false + } + return mp.Stats(), true +} + // GossipStats returns a snapshot of gossipsub block-ingestor counters // and true when gossipsub head-tracking is active. Returns (zero, false) // when running on the polling Sync alone. Useful for verifying the diff --git a/pkg/daemon/start.go b/pkg/daemon/start.go index 85a30f2..4d9b70a 100644 --- a/pkg/daemon/start.go +++ b/pkg/daemon/start.go @@ -688,8 +688,20 @@ func (d *Daemon) startGossipHead(ctx context.Context, store *hstore.Store, src b // Best-effort: a mpool wiring failure must not sink head-tracking, it // just leaves eth_sendRawTransaction on the bridge fallback. if chainAPI != nil { + // #119: derive the durable persist path. Default is + // //mpool/pending.jsonl unless the caller passed + // an override, or MpoolPersistDisabled=true (empty path = memory-only). + persistPath := "" + if !d.cfg.MpoolPersistDisabled { + if d.cfg.MpoolPersistPath != "" { + persistPath = d.cfg.MpoolPersistPath + } else { + persistPath = filepath.Join(d.cfg.DataDir, network.String(), "mpool", "pending.jsonl") + } + } mp, mperr := mpool.New(ctx, host.PubSub, mpool.Config{ - Topic: network.GossipTopicMessages(), + Topic: network.GossipTopicMessages(), + PersistPath: persistPath, }) if mperr != nil { log.Warnw("mpool publisher unavailable; eth_sendRawTransaction stays on bridge", "err", mperr) @@ -698,7 +710,7 @@ func (d *Daemon) startGossipHead(ctx context.Context, store *hstore.Store, src b d.mu.Lock() d.mpool = mp d.mu.Unlock() - log.Infow("gossipsub mempool publisher wired", "topic", network.GossipTopicMessages()) + log.Infow("gossipsub mempool publisher wired", "topic", network.GossipTopicMessages(), "persistPath", persistPath, "restored", mp.Stats().Restored) // lantern#47: drive the mpool's pending -> confirm -> rebroadcast // loop on every head advance. StateSearchMsg is local + zero-Glif