refactor(networkx): make the recovery reload a process-local flag

lightrag/kg/faiss_impl.py

@@ -359,6 +359,12 @@ def __post_init__(self):
         # How many times the file channel caught a commit the flag channel
         # never announced, so a deployment can tell whether the
         # lost-notification window in #3854 actually occurs.
+        # The on-disk state already counted as a lost notification. A load
+        # that raises leaves _loaded_fingerprint in place, so the same peer
+        # commit is re-detected by every later call; without this it would
+        # also be re-counted, without bound. See
+        # file_fingerprint.counts_as_a_new_lost_notification.
+        self._counted_peer_fingerprint = None
         self._missed_notification_reloads = 0
 
         # Minimal pending area for deferred embedding: custom-id -> _PendingFaissDoc.
@@ -451,7 +457,26 @@ def _adopt_fingerprint(
         self, fingerprint: file_fingerprint.Fingerprint | object
     ) -> None:
         """Record ``fingerprint`` as the file pair this process now holds."""
-        self._loaded_fingerprint = file_fingerprint.adopted(fingerprint)
+        adopted = file_fingerprint.adopted(fingerprint)
+        self._loaded_fingerprint = adopted
+        # The dedupe marker's job ends here -- but ONLY if a concrete state
+        # was recorded. It exists to stop a detection being re-counted while
+        # the reload that should discharge it keeps failing, and a landed
+        # reload normally ends that: ``_loaded_fingerprint`` IS this state
+        # from here, so any later divergence is genuinely new. Keeping it
+        # past that point would suppress a state that RECURS -- a peer drop,
+        # a notified recreation, then a second drop whose notification is
+        # lost, all sharing the "absent" fingerprint, which is a real second
+        # loss and not the same-tick collision residue.
+        #
+        # ``adopted(UNREADABLE)`` is ``None``, which is not a state: it means
+        # "nothing recorded", and ``peer_commit_detected`` reports a change
+        # against it for ANY state. Clearing on that would forget which
+        # commit was already counted and count the same one again on the next
+        # call. The post-drop fingerprint is ``(None,)`` -- a real, concrete
+        # state -- so a drop still clears.
+        if adopted is not None:
+            self._counted_peer_fingerprint = None
 
     def _record_fingerprint(self) -> None:
         """Adopt the files currently on disk without reloading from them.
@@ -502,22 +527,40 @@ def _reload_index_from_disk_locked(self, *, for_write: bool = False) -> bool:
                 logger.warning(log_message)
             else:
                 logger.info(log_message)
-        else:
+        # Sampled BEFORE the read, never after -- see ``kg.file_fingerprint``.
+        # Hoisted above the logging so the counting below can deduplicate on
+        # the very sample this reload will adopt, at no extra stat.
+        fingerprint = self._stat_fingerprint()
+
+        if not notified:
             # The lost-notification case the file channel exists for. Always a
             # warning, on the read path too: unlike a notified reload this one
             # says a publication failed somewhere.
-            self._missed_notification_reloads += 1
-            logger.warning(
-                f"[{self.workspace}] Process {os.getpid()} FAISS reloading "
-                f"{self.namespace}: {self._faiss_index_file} is not the file "
-                "pair this process loaded and no reload notification arrived "
-                "for it, so a notification was lost. Recovering through the "
-                f"file channel (occurrence #{self._missed_notification_reloads} "
-                "in this process)."
-            )
-
-        # Sampled BEFORE the read, never after -- see ``kg.file_fingerprint``.
-        fingerprint = self._stat_fingerprint()
+            #
+            # Counted once per on-disk state, not once per detection: a load
+            # that raises below leaves _loaded_fingerprint in place, so this
+            # same commit is re-detected by every later call. See
+            # ``file_fingerprint.counts_as_a_new_lost_notification``.
+            if file_fingerprint.counts_as_a_new_lost_notification(
+                fingerprint, self._counted_peer_fingerprint
+            ):
+                self._counted_peer_fingerprint = file_fingerprint.adopted(fingerprint)
+                self._missed_notification_reloads += 1
+                logger.warning(
+                    f"[{self.workspace}] Process {os.getpid()} FAISS reloading "
+                    f"{self.namespace}: {self._faiss_index_file} is not the file "
+                    "pair this process loaded and no reload notification arrived "
+                    "for it, so a notification was lost. Recovering through the "
+                    f"file channel (occurrence #{self._missed_notification_reloads} "
+                    "in this process)."
+                )
+            else:
+                logger.debug(
+                    f"[{self.workspace}] The peer commit to "
+                    f"{self._faiss_index_file} is the pair already counted, or "
+                    "its stat failed; this is a retry of a reload that did not "
+                    "land, not a second lost notification."
+                )
         self._index = faiss.IndexFlatIP(self._dim)
         self._id_to_meta = {}
         self._load_faiss_index()

lightrag/kg/file_fingerprint.py

@@ -18,6 +18,81 @@
 two commits land inside one filesystem timestamp tick with an identical size,
 which needs a healthy, fast-committing system -- exactly when the flag works.
 
+**The deferred remedy for the tick collision**, recorded here so it is not
+rediscovered from scratch: make the writer guarantee mtime monotonicity --
+``stat`` the target before the commit and, if ``os.replace`` did not advance
+its mtime, bump it with ``os.utime``. That would make the file channel exact
+on its own. It is deliberately NOT done, because the bump has no good value on
+a coarse filesystem: ``+1 ns`` is truncated away on a 1 s (ext3, HFS+) or 2 s
+(FAT) granularity, and a whole-granule bump produces user-visible future
+timestamps. It costs one extra ``stat`` per commit plus a rare ``utime``, so
+cost is not the objection. Do it only if the ``_missed_notification_reloads``
+counters that each storage logs ever show this window occurring in a real
+deployment -- those counters are the evidence this decision waits on, which is
+why the next section exists.
+
+``_missed_notification_reloads``: one increment per peer commit
+---------------------------------------------------------------
+
+Each storage keeps this counter, and its contract is exactly:
+
+    **it increments once per peer commit that no notification announced --
+    not per detection of one, and not per attempt to reload out of one.**
+
+It reads like a log line and is not one. It is the instrument the ``os.utime``
+decision above waits on, so a bias in it is not cosmetic: it silently decides
+whether that work is ever judged necessary. Both directions are defects, and
+review found this counter wrong in five distinct ways, each fixed in a
+different place. They are indexed here because no single site shows the whole
+contract, and the next person to touch any one of them will be looking at a
+fragment:
+
+1. **Do not arm a cross-process channel for a process-local fact.**
+   ``NetworkXStorage`` recovers from a failed save by discarding its
+   unpersisted graph. Arming that through ``_loaded_fingerprint`` made the
+   file channel report a peer commit that never happened. It uses a
+   process-local ``_recovery_reload_pending`` bool instead -- see that class's
+   *Recovery reload*. (Overcount.)
+2. **Classify before a reload that discharges several conditions at once.**
+   That same recovery flag outranks both channels, and one reload satisfies
+   all of them, so a peer commit arriving while recovery is pending would be
+   handled and never counted. Both recovery branches count first, then
+   reload. (Undercount.)
+3. **Count states, not attempts.** A reload that raises leaves the reader's
+   recorded fingerprint untouched, so the same commit is re-detected by every
+   later call. :func:`counts_as_a_new_lost_notification` plus each storage's
+   ``_counted_peer_fingerprint`` makes it once-per-state. (Overcount, and it
+   was unbounded.)
+4. **End that marker at the reload it was protecting -- and not before.**
+   Kept longer, it suppresses a state that RECURS -- a drop, a notified
+   recreation, a second drop -- which is a real second loss and not the
+   tick-collision residue. Cleared in each storage's ``_adopt_fingerprint``,
+   the single point a new state is recorded and one reached only after a load
+   or commit landed. **Only when that adoption records a CONCRETE state**,
+   though: ``adopted(UNREADABLE)`` is ``None``, which means "nothing
+   recorded" and against which any state reads as a change, so clearing there
+   forgets which commit was counted and counts it again. The post-drop
+   fingerprint is ``(None,)``, a real state, so a drop still clears. (Both
+   directions: undercount if kept too long, double-count if dropped too
+   early.)
+5. **Once a call is committed to adopting, it must not observe the file
+   again.** Every step from there -- deciding there is a divergence, counting
+   it, adopting the new state -- runs on ONE sample. A second observation can
+   come back ``UNREADABLE`` while the first succeeded, and then its step is
+   skipped while the adoption still happens on the good sample, erasing the
+   divergence a later call would have counted. So the counting callers pass
+   their sample to :func:`divergence_detected`, to
+   :func:`counts_as_a_new_lost_notification` and to their reload alike.
+   An observation *before* that point is fine and the vector backends use one:
+   theirs gates the whole function and returns early, adopting nothing, so a
+   failure there costs a retry rather than the event. Found twice, both after
+   the commit point: first the count-vs-adopt pair, then the divergence test
+   that was still re-observing. (Loss, not merely undercount.)
+
+What the counter still cannot see is the tick collision itself -- two commits
+sharing one ``(st_mtime_ns, st_size)`` -- which is the residue the remedy above
+would remove. That is the one blind spot by design; the five above were not.
+
 The mechanism lives here, once, because its hazards are in the details rather
 than the shape, and three copies of them is how it rots:
 
@@ -125,6 +200,46 @@ def adopted(sampled: Fingerprint | object) -> Fingerprint | None:
     return None if sampled is UNREADABLE else sampled  # type: ignore[return-value]
 
 
+def counts_as_a_new_lost_notification(
+    sampled: Fingerprint | object, already_counted: Fingerprint | None
+) -> bool:
+    """Whether this detection is a NEW lost notification, not a re-detection.
+
+    Every storage here keeps a ``_missed_notification_reloads`` counter, and
+    the module docstring designates those counters as the evidence the
+    writer-side ``os.utime`` remedy waits on. That only holds if they count
+    **peer commits**, and detection alone does not: a reload that raises
+    leaves the reader's recorded fingerprint untouched, so the same peer
+    commit is re-detected by every later call. Counted at each detection, one
+    commit inflates the counter without bound -- and a file that stays
+    unreadable for a while is not exotic, since that is what a sick storage
+    looks like.
+
+    So each storage remembers the state it last counted and passes it here.
+    A genuinely later commit has a different fingerprint and counts again;
+    two commits inside one timestamp tick with an identical size do not, which
+    is the collision residue this fence already documents, inherited rather
+    than newly introduced.
+
+    ``UNREADABLE`` counts nothing: it cannot say WHICH state it would be
+    counting, so the count could neither be deduplicated nor trusted. The next
+    call counts it if the ``stat`` works by then -- but ONLY because the
+    caller feeds this same sample to its reload, so an unreadable one adopts
+    no fingerprint and leaves the divergence standing. A caller that sampled
+    here and let its reload sample independently would lose the event for
+    good: this would skip the count while that sample succeeded and adopted
+    the peer state. Count and adopt from one observation.
+
+    Counting at detection rather than after a successful reload is deliberate:
+    the window occurred whether or not this process could reload out of it,
+    and a file that never becomes readable would otherwise erase the evidence
+    entirely.
+    """
+    if sampled is UNREADABLE:
+        return False
+    return sampled != already_counted
+
+
 def peer_commit_detected(
     paths: Sequence[str], recorded: Fingerprint | None, *, workspace: str
 ) -> bool:
@@ -151,7 +266,33 @@ def peer_commit_detected(
     """
     if not fence_enabled():
         return False
-    sampled = sample(paths, workspace=workspace)
+    return divergence_detected(
+        sample(paths, workspace=workspace), recorded, paths=paths, workspace=workspace
+    )
+
+
+def divergence_detected(
+    sampled: Fingerprint | object,
+    recorded: Fingerprint | None,
+    *,
+    paths: Sequence[str],
+    workspace: str,
+) -> bool:
+    """:func:`peer_commit_detected`'s decision, from a sample already taken.
+
+    For the caller that will DECIDE, COUNT and ADOPT within one call: all
+    three must come from **one observation**. Taking a fresh ``stat`` for the
+    decision lets it fail while the caller's sample succeeded, and then the
+    count is skipped while the reload adopts that good sample -- erasing the
+    divergence that would have let a later call count the event. The
+    one-observation rule in :func:`counts_as_a_new_lost_notification` covers
+    counting and adoption; this covers the third participant.
+
+    A caller with nothing else to do with the sample should use
+    :func:`peer_commit_detected`, which takes one for itself.
+    """
+    if not fence_enabled():
+        return False
     if sampled is UNREADABLE:
         return False
     if sampled == recorded:
@@ -203,6 +344,17 @@ def publication_complete(sampled: Fingerprint) -> bool:
 
     Everything absent is complete (the post-``drop`` state, which peers must
     be able to converge on). A present marker with any file missing is not.
+
+    **This test is timestamp-based because the formats give it nothing else.**
+    The deferred remedy, recorded so it is not rediscovered: put an explicit
+    generation counter in the metadata each file carries (FAISS's
+    ``meta.json``), or publish the set atomically -- stage a whole generation
+    in a directory and rename that directory into place -- and completeness
+    becomes a comparison of equal generation numbers, independent of the
+    filesystem clock and of the strictness argument above. It is not done here
+    because it is a storage-format change, and these multi-file backends are
+    development and test storage today. It is the right answer if they ever
+    become production storage.
     """
     *committed, marker = sampled
     if marker is None: