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 



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).



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.



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