Add the keygrabber collector: config, scheduling and the daemon

[mikelangmayr]

Makes the keygrabber run. Builds on the sink layer from #32 with config parsing, the read path, and the daemon that drives them.

• Add KeygrabberDaemon: a scheduler thread working a due-time heap, a bounded worker pool reading through one Client, and a single writer thread that owns every sink call, so no reader thread touches the backend
• Add Collection, which resolves a peer's keywords from one keys.list and reads them as one keys.read per tick. Selection is local, so a collection with several patterns still costs one request
• Detect bulk-read support from the services field rather than probing, and fall back to per-keyword reads for a peer on an older libby. The fallback abandons the rest of a tick after the first timeout, since paying timeout_s per keyword would overrun the interval many times over
• Stamp every value in a tick with one keygrabber-side timestamp, keeping clock skew between daemon hosts out of the data
• Add config parsing with validation: collection names must match [a-z0-9_]+ so they can become control keyword prefixes, peer must be <group>.<daemon>, and cadences must be numbers
• Reject interval_s at or below 1.5 x timeout_s. bamboo waits timeout_s for the ACK and timeout_s / 2 more for the reply, so one unanswered read can occupy a worker for one and a half timeouts
• Skip a tick whose predecessor is still running and count it, so a wedged peer cannot accumulate overlapping reads. The scheduler also never schedules into the past, so a long stall cannot queue a burst of catch-up ticks that can only skip
• Exclude uptime and lasterror by default: uptime changes every second and says nothing a timestamp does not, and lasterror is null most of the time. Naming either explicitly opts it back in
• Refuse an inline sink.token, requiring token_env instead, and validate every sink field before importing the backend so a malformed config reports the same way whether or not the extra is installed
• Add a keygrabber console entry point, with exits confined to main.py so the daemon stays importable
• Document it: a new docs/source/keygrabber.md covering running it, the config schema, how it reads and how it stores, wired into the toctree, the API reference and the README
• Run the end-to-end daemon tests on both transports, six cases each, via shared base classes. The ZMQ cases need no broker and always run; the RabbitMQ cases skip without one. RabbitMQ is the default transport and the only one where the worker pool issues concurrent requests over a single pika connection, so testing it is the point rather than a formality
• 198 tests pass with and without the influx extra, pylint 10.00/10, -W docs build verified locally

One thing a reviewer may want to push on: KeygrabberDaemon carries twelve attributes and has a too-many-instance-attributes disable. Extracting the scheduler into its own class would fix that and would also let the due-time ordering and the skip rule be unit-tested against a fake clock, which today are only covered end to end. That extraction is in progress on the follow-up branch, since reload needs the same boundary.

The control keywords are not here: enabled, the counters, reload and the per-collection interval land next, so for now the daemon is observable only through its logs.

[prkrtg]

What happens if submit fails? (basically the pool exists but already shut down) Should this be wrapped in try/except?

try:
            pool.submit(self._run_tick, collection)
        except RuntimeError:
            with self._in_flight_lock:
                self._in_flight.discard(name)

[mikelangmayr]

Good catch, it can fail and the consequence is worse than a crash: the claim was never released, so that collection would look permanently busy and every later tick would be skipped. _dispatch now returns False for both a shut-down pool and a None one, and _submit gives the claim back. Added a regression test that shuts the pool down while leaving it non-None, which is the exact race.

[prkrtg]

Should the writer thread do its own draining? When it sees _halt, it stops taking new work, drains what's left, and exits. Then on_stop just waits for it and only calls close() if exited

[mikelangmayr]

Yes, and that is better than what I had. Draining from on_stop could put two threads in one sink at once, since the join had a timeout and RetryingWriter is not thread-safe. The writer now drains at the end of its own loop, on_stop waits out DRAIN_DEADLINE_S plus a margin, and only calls close() if the thread actually exited; otherwise it logs and leaves the sink to process exit rather than closing underneath it.