Make the cursor mean nothing when the addresses under it change

A cursor is only meaningful for the trust root it was read with. Change the
factory and every PoolCreated it ever emitted is already behind the cursor, so
the new address is never seen and the old rows stay — the store keeps answering
for a book that is not on this chain.

Zoo mainnet sat that way. Its declared V3 factory held no code; the real one, at
the address Lux uses, has twelve pools. Correcting the address alone would have
changed nothing, because the cursor was at the tip.

So the source is part of the store's identity, exactly as the genesis hash is
part of the chain's, and a change to either rewinds. An unrecorded source is
unknown rather than different: every store written before this has none, and
reading that as a change would re-read a million blocks on Lux to rebuild rows
that are already right.

Co-authored-by: Hanzo Dev <[email protected]>

indexer/indexer.go

@@ -184,6 +184,33 @@ func NewWithConfig(cfg Config, store *storage.Store) *Indexer {
 		},
 	}
 	idx.genesisHashFn = idx.fetchGenesisHash
+
+	// A cursor only means anything for the addresses it was read with. Change
+	// the factory and every event it ever emitted is already behind the cursor,
+	// so the new address is never seen and the old rows stay — the store keeps
+	// answering for a book that is not on this chain. Zoo mainnet sat that way:
+	// its declared factory held no code, the real one had twelve pools, and the
+	// correction alone would have changed nothing because the cursor was at the
+	// tip. So the source is part of the store's identity, exactly as the genesis
+	// hash is part of the chain's, and a change to either rewinds.
+	//
+	// An UNRECORDED source is unknown, not a mismatch. Every store written before
+	// this existed has none, and reading that as "different" would re-read every
+	// chain from genesis the first time this ships — a million blocks on Lux, to
+	// rebuild rows that are already right. Unknown is its own state: adopt the
+	// current set as the store's and drop nothing.
+	fp := strings.ToLower(strings.Join([]string{idx.factoryV2, idx.factoryV3, idx.poolManager, idx.native}, "|"))
+	switch was := store.Source(); {
+	case was == "":
+		store.SetSource(fp)
+	case was != fp:
+		idx.logf("[indexer] source changed (%s -> %s) — re-reading from %d", short(was), short(fp), cfg.StartBlock)
+		if err := store.DropDerived(); err != nil {
+			idx.logf("[indexer] source changed but the old rows would not drop: %v", err)
+		}
+		store.SetSource(fp)
+	}
+
 	idx.lastBlock = idx.store.GetLastBlock()
 	if idx.lastBlock < cfg.StartBlock {
 		idx.lastBlock = cfg.StartBlock
@@ -1085,3 +1112,18 @@ func (idx *Indexer) isKnownPool(addr string) bool {
 	p, _ := idx.store.GetPool(nil, strings.ToLower(addr))
 	return p != nil
 }
+
+// short renders a source fingerprint for a log line: the four addresses it
+// joins are 170 characters, and what a reader needs is only that it changed.
+func short(fp string) string {
+	if fp == "" {
+		return "(unrecorded)"
+	}
+	parts := strings.Split(fp, "|")
+	for i, p := range parts {
+		if len(p) > 10 {
+			parts[i] = p[:10]
+		}
+	}
+	return strings.Join(parts, "|")
+}

storage/store_sqlite.go

@@ -98,6 +98,44 @@ func (s *Store) SetLastBlock(block uint64) {
 	s.db.Exec("INSERT OR REPLACE INTO meta(key, value) VALUES('lastBlock', ?)", strconv.FormatUint(block, 10))
 }
 
+// SetSource records which addresses the rows below the cursor were built from.
+// Called once the indexer has committed to a set, so a later start can tell
+// whether the cursor it inherits describes the same book.
+func (s *Store) SetSource(fingerprint string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.db.Exec("INSERT OR REPLACE INTO meta(key, value) VALUES('source', ?)", fingerprint)
+}
+
+// Source returns the fingerprint the rows were built from, or "" for a store
+// written before this was recorded — which is indistinguishable from a store
+// built by another set, and is treated as one.
+func (s *Store) Source() string {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	var v string
+	if err := s.db.QueryRow("SELECT value FROM meta WHERE key='source'").Scan(&v); err != nil {
+		return ""
+	}
+	return v
+}
+
+// DropDerived removes everything read off the chain, leaving the file and its
+// schema. Used when the addresses changed under a store: those rows describe a
+// different book, and mixing them with the new one gives an answer that was
+// never true anywhere.
+func (s *Store) DropDerived() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, t := range []string{"factories", "bundles", "tokens", "pools", "swaps", "entities"} {
+		if _, err := s.db.Exec("DELETE FROM " + t); err != nil {
+			return err
+		}
+	}
+	_, err := s.db.Exec("DELETE FROM meta WHERE key='lastBlock'")
+	return err
+}
+
 // GetLastBlock returns the indexer's last processed block.
 func (s *Store) GetLastBlock() uint64 {
 	s.mu.RLock()