Entity indexer concurrency¶

Use this reference when multiple threads or processes can access zettelforge.entity_indexer.EntityIndexer. ZettelForge 2.8.0 serializes in-process mutations and creates atomic file snapshots. Concurrent lookup safety is still in progress: search_entities() and stats() can fail while another thread mutates the index.

Module: zettelforge.entity_indexer
Verified release: ZettelForge 2.8.0

Prerequisites¶

Install ZettelForge 2.8.0 on Python 3.10 or newer. Start with the Quickstart if you do not have a working installation.
Treat one EntityIndexer instance as shared mutable state when threads use it.
Coordinate separate processes outside ZettelForge. Its file lock does not serialize processes that write the same index path.

Concurrent lookup safety is in progress

Do not call search_entities(), stats(), or build() at the same time as a mutation. A release-tree stress check reproduced RuntimeError: dictionary changed size during iteration in both lookup methods. get_note_ids() is also unlocked and does not promise a stable snapshot.

In-memory structure¶

The index maps an entity type to normalized entity values and then to note IDs:

entity_type -> entity_value -> set[note_id]

Construction creates the 19 built-in type buckets. remove_note() preserves those buckets when their last value disappears. add_note() also accepts a new type key, so the key set can grow beyond the 19 built-in types during the instance lifetime. build() resets the index to the built-in buckets before it rebuilds the contents.

Thread-safety contract¶

EntityIndexer uses one threading.RLock, stored as _flush_lock, for mutation, snapshot creation, loading, timer coordination, and dirty-state transitions.

Operation	Locking in 2.8.0	Concurrent-use contract
`load()`	Holds `_flush_lock` while it reads and replaces known buckets	Serialized with locked mutations on the same instance
`save()`	Copies the index under `_flush_lock`; writes the copy after releasing it	Produces a stable point-in-time snapshot
`add_note()`	Holds `_flush_lock` for the complete mutation	Serialized with `load()`, `save()` snapshots, `remove_note()`, and flushes
`remove_note()`	Holds `_flush_lock` for the complete mutation	Serialized with `load()`, `save()` snapshots, `add_note()`, and flushes
`_schedule_flush()`	Holds `_flush_lock` while it inspects and creates the timer	Starts at most one live timer per instance
`_flush_sync()`	Holds `_flush_lock` across dirty check, `save()`, and dirty clear	Serializes a scheduled flush with mutations
`get_note_ids()`	Unlocked	No stable-snapshot guarantee during mutation
`search_entities()`	Unlocked	Unsafe during mutation; reproduced `RuntimeError`
`stats()`	Unlocked	Unsafe during mutation; reproduced `RuntimeError`
`build()`	Does not lock the rebuild as one operation	Run only with exclusive access to the instance

Why the lock is reentrant¶

_flush_sync() acquires _flush_lock and calls save(). save() acquires the same lock to create its snapshot. A non-reentrant threading.Lock would deadlock on that nested acquisition; threading.RLock lets the flush thread acquire the same lock again.

What a saved snapshot means¶

save() copies every mapping while it holds _flush_lock. A concurrent add_note() or remove_note() cannot change the index during that copy. The method then releases the lock and writes the captured copy, so the file represents one complete point in time rather than every mutation that finishes before the disk write returns.

The background flush preserves later mutations: they set _dirty and schedule another flush when no live timer exists.

Background flush lifecycle¶

add_note() and remove_note() mark the index dirty, then schedule a five-second daemon timer when no timer is alive. Mutations that arrive before that timer fires join the same batch. The timer calls _flush_sync(), which saves under the reentrant lock and clears _dirty after the save succeeds.

The constructor also registers _flush_sync() with atexit for a final flush on normal interpreter shutdown. A daemon timer does not keep a process alive.

build() cancels a live timer, saves synchronously, and clears _dirty before it returns. The rebuild itself lacks an operation-wide lock, so give build() exclusive access to the instance.

File persistence contract¶

save() follows this sequence:

Create a stable in-memory copy under _flush_lock.
Create a temporary file in the target directory.
Lock that temporary file with fcntl.flock(LOCK_EX).
Write JSON, flush it, and call os.fsync() on the file.
Unlock the temporary file and replace the target with os.replace().

On POSIX, the final replace gives readers atomic target-file visibility: they open the old complete file or the new complete file, not a partially written target. If serialization fails, save() deletes the temporary file and raises the exception.

Cross-process limit¶

Each writer creates and locks its own temporary file. Those locks do not serialize independent processes. When two processes save the same target path, each can complete its own snapshot and the last successful os.replace() wins. Use an external process lock when more than one process can write the same entity_index.json.

Public API¶

`EntityIndexer`¶

Method	Signature	Result
Constructor	`EntityIndexer(index_path: str \\| None = None)`	Loads the default or supplied JSON index path
`load`	`() -> bool`	`True` after a successful load; `False` when the path is absent or loading fails
`save`	`() -> None`	Writes one atomic snapshot or raises the write error
`add_note`	`(note_id: str, entities: dict[str, list[str]]) -> None`	Adds normalized values and schedules a flush
`remove_note`	`(note_id: str) -> None`	Removes the note ID, prunes empty values, and schedules a flush
`get_note_ids`	`(entity_type: str, entity_value: str) -> list[str]`	Returns matching note IDs or an empty list
`search_entities`	`(query: str, limit: int = 10) -> dict[str, list[str]]`	Returns case-insensitive prefix matches, limited per type
`stats`	`() -> dict`	Returns unique-entity and note-mapping counts per type
`build`	`() -> dict`	Rebuilds from stored notes and returns count plus statistics

The following executed example shows the result shapes. It uses a temporary index and an explicit save():

from pathlib import Path
from tempfile import TemporaryDirectory

from zettelforge.entity_indexer import EntityIndexer

with TemporaryDirectory() as directory:
    index_path = Path(directory) / "entity_index.json"
    indexer = EntityIndexer(index_path=str(index_path))
    indexer.add_note(
        "note-1",
        {"actor": ["APT28"], "tool": ["Cobalt Strike"]},
    )
    indexer.save()

    assert indexer.get_note_ids("actor", "apt28") == ["note-1"]
    assert indexer.search_entities("cobalt") == {
        "tool": ["cobalt strike"],
    }
    print(indexer.stats()["actor"])

Output:

{'unique_entities': 1, 'total_mappings': 1}

Verification boundary¶

The ZettelForge 2.8.0 release passed 35 focused entity-indexer and conversational- entity tests during this audit. A separate bounded stress check reproduced the two concurrent lookup failures described above. The official regression suite covers atomic save, temporary-file cleanup, built-in bucket preservation, and concurrent save-plus-add behavior; it does not assert concurrent-read safety.