Class: Pikuri::VectorDb::Watcher

Inherits:
Object
  • Object
show all
Defined in:
lib/pikuri/vector_db/watcher.rb

Overview

The auto-watch daemon: keep a persistent index in sync with a corpus on disk without the user ever calling vectordb_reindex. A thin layer over the listen gem driving the Indexer's per-file entry points (Indexer#reindex_file! / Indexer#remove_file!) from filesystem events, plus a boot reconciliation sweep (Indexer#reconcile_plan).

Host-owned, not agent-owned

A web app has many Agents but one corpus and one Watcher, so its lifetime is the host *process*. The host constructs it once around the same Indexer every agent's Extension shares:

server = Pikuri::VectorDb::Server::Chroma.ensure_running
backend = server.client(collection: 'notes')
ext = Pikuri::VectorDb::Extension.new(backend: backend, source: '~/notes')
# ... build agents with `ext` ...
watcher = Pikuri::VectorDb::Watcher.new(indexer: ext.indexer).start

The Extension indexes nothing on its own, so the Watcher is the population path: its boot sweep fills an empty backend, uniform across backends.

Single work queue, last-intent-per-path

The boot sweep and live events must not race, so both funnel into one @pending map (path → +:reindex+/+:remove+) drained by one worker thread:

  • Rapid edits to one file collapse to a single reindex (one intent per path); +listen+'s latency window coalesces most before it reaches us.
  • Order can't corrupt content: #process resolves the file from disk at process time, and reindexing a vanished file is itself a removal — so a stale :reindex after a delete self-corrects rather than writing old bytes.

The boot sweep is enqueued after the listener is subscribed, so a mid-sweep change lands in the same queue rather than being lost.

Teardown drains the in-flight job

The Watcher registers with Finalizers last (after the host's Server::Chroma), so LIFO closes it first. #close stops the listener, then joins the worker, which finishes its in-flight job and exits — the load-bearing ordering that guarantees no Backend#replace_source is still writing when the container goes down. Still-queued jobs are abandoned; the next boot's sweep catches them (the downtime-gap guarantee).

What it deliberately does not do

  • No model/parameter awareness. The change signal is the byte hash (Indexer#file_hash), so swapping the embedder/chunker config does not trigger reindex — recourse is a manual vectordb_reindex (see Extension).
  • Symlinks are never indexed. A live event reached through a symlink below the watch root (leaf or ancestor) is dropped, matching Indexer#list_files so watch and walk index the same set. Strict on purpose: a privacy boundary, and it closes an off-tree divergence — a symlink target changing outside the tree fires no event, so following it would leave stale chunks.
  • No backpressure. Re-embedding competes with the chat model on a shared llama-server; vectordb targets low-churn corpora, so a burst is tolerated rather than throttled.

Constant Summary collapse

LOGGER = 
Pikuri.logger_for('VectorDb::Watcher')

Instance Method Summary collapse

Constructor Details

#initialize(indexer:, latency: 10.0, listen_factory: nil) ⇒ Watcher

Parameters:

  • indexer (Indexer) —

    the indexer to drive — supplies the watch Indexer#root, the source-vs-directory shape via Indexer#source, and the per-file + reconcile entry points.

  • latency (Float) (defaults to: 10.0) —

    listen debounce window in seconds: how long filesystem events are batched before the callback fires. The corpus is expected to change rarely, so a coarse value (default 10s) coalesces editor save-storms without any cost.

  • listen_factory (#call, nil) (defaults to: nil) —

    dependency-injection seam for tests: call(dir) { |modified, added, removed| ... } returns an object responding to #start / #stop. nil uses the real Listen.to (required lazily — see the gemspec note).



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/pikuri/vector_db/watcher.rb', line 85

def initialize(indexer:, latency: 10.0, listen_factory: nil)
  @indexer = indexer
  @latency = latency
  @listen_factory = listen_factory || lambda do |dir, &blk|
    require 'listen'
    Listen.to(dir, latency: @latency, &blk)
  end

  # Resolve the watch shape once, here, rather than re-stat'ing
  # per event. {Indexer#root} is the source for a directory tree
  # and the *parent* for a single file, so +source != watch_dir+
  # distinguishes the two without touching the filesystem — and
  # stays correct even if the single watched file is later
  # deleted (a +source.file?+ check would flip to directory mode
  # and start indexing its siblings).
  @source    = @indexer.source.expand_path
  @watch_dir = @indexer.root.expand_path
  @single_file = @source != @watch_dir

  @pending  = {} # path (String) → :reindex / :remove
  @monitor  = Monitor.new
  @cond     = @monitor.new_cond
  @closing  = false
  @started  = false
  @closed   = false
  @worker   = nil
  @listener = nil
  @finalizer_handle = nil
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.

Stop watching and tear down cleanly. Stops the listener so no new events arrive, then joins the worker — which finishes its in-flight job and exits, abandoning any still-queued work. Idempotent and safe at process exit (the host may have called it already, or only the at_exit finalizer fires).



139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/pikuri/vector_db/watcher.rb', line 139

def close
  return if @closed

  @closed = true
  @listener&.stop
  @monitor.synchronize do
    @closing = true
    @cond.broadcast
  end
  @worker&.join
  Pikuri::Finalizers.unregister(@finalizer_handle) if @finalizer_handle
  nil
end

#drain! ⇒ void

This method returns an undefined value.

Process every currently-queued job synchronously and return. The worker thread calls this implicitly via its blocking loop; tests call it directly to drain without a thread. Does not wait for new work and does not consult the closing flag.



182
183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/pikuri/vector_db/watcher.rb', line 182

def drain!
  loop do
    job = @monitor.synchronize do
      next nil if @pending.empty?

      key = @pending.keys.first
      [key, @pending.delete(key)]
    end
    break if job.nil?

    process(*job)
  end
  nil
end

#process_event(modified, added, removed) ⇒ void

This method returns an undefined value.

The listen callback body, also the test seam. Modified/added → :reindex, removed → :remove; non-indexable paths (denylisted, dot-files, outside the root) and symlinked paths are dropped. The symlink check guards only :reindex (it +lstat+s, which a removed path lacks, and a :remove for a never-indexed symlink source is a no-op).

Parameters:

  • modified (Array<String>) —

    absolute paths.

  • added (Array<String>) —

    absolute paths.

  • removed (Array<String>) —

    absolute paths.



163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/pikuri/vector_db/watcher.rb', line 163

def process_event(modified, added, removed)
  (Array(modified) + Array(added)).each do |path|
    next unless indexable?(path)
    next if symlinked_below_root?(path)

    enqueue(:reindex, path)
  end
  Array(removed).each do |path|
    enqueue(:remove, path) if indexable?(path)
  end
  nil
end

#start ⇒ self

Subscribe to filesystem events, spawn the worker thread (which runs the boot reconcile sweep then drains events), and register for teardown. Idempotent guard against a double start.

Returns:

  • (self) 


120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/pikuri/vector_db/watcher.rb', line 120

def start
  raise 'Watcher already started' if @started

  @started = true
  subscribe
  @worker = Thread.new { work_loop }
  # Register LAST so LIFO teardown closes the Watcher before the
  # Server::Chroma the host registered earlier — see the class yardoc.
  @finalizer_handle = Pikuri::Finalizers.register(self)
  self
end