david
bb852faa27
Implements the top findings from the codebase review, verified with tests and live CLI/MCP checks. **Gossip integrity** - Push validation: 4 MiB body cap, 1000-row batch cap; rows claiming the local node id (vector-poisoning), empty node ids, and negative HCLs rejected (internal/watch/gossip.go, internal/db/gossip.go) - Reconcile-on-pull: Run returns the pulled count, syncGossip rebuilds derived state when > 0 — entry-count comparison could never fire, so synced observations never materialized into searchable entries **Data-layer safety** - HLC resumed from MAX(hcl) at Open (hlc.SeekTo): a restart with a regressed wall clock cannot reissue values the (node_id, hcl) locator and pull cursors depend on - Writer serialization: _txlock=immediate DSN + SetMaxOpenConns(1) + per-KnoxDB mutex around RecordObservation's check-then-insert dedup (closes duplicate-row race) **Watch daemon** - Ticker guard flags now atomic.Bool (was a cross-goroutine data race) - Trailing-edge per-path debounce (timer-based, pruned on fire/delete) - Recursive watches (startup tree walk + watcher.Add on dir Create); Rename re-ingests, Remove cancels pending ingests **MCP + CLI** - Strict arg validation, no silent clamping: thread_id 0 errors instead of renaming thread #1; empty knox_thread_link {} errors instead of false success; thread existence checked before writes; golden-thread tool nil-safe - --page 0 errors instead of panicking; query/recent pagination actually pages (page x limit) **Tests** (new internal/hlc and internal/db packages): SeekTo monotonicity, concurrent dedup race, push validation, reopen HCL monotonicity, batch caps, self-spoof rejection, idempotency on observation counts. Verified: go build, go vet, full suite with -race, live MCP stdio transcripts against a scratch DB. Reviewed-on: #3 Co-authored-by: David Gwilliam <dhgwilliam@gmail.com> Co-committed-by: David Gwilliam <dhgwilliam@gmail.com>
78 lines
2.5 KiB
Go
78 lines
2.5 KiB
Go
// Package hlc implements a Hybrid Logical Clock for ordering observations.
|
|
//
|
|
// A node's HLC is monotonic even when wall clocks jump (NTP correction, suspend).
|
|
// Values are packed into an int64: high bits = wall-clock milliseconds, low bits
|
|
// = per-millisecond sequence. Lexicographic comparison of the packed value is a
|
|
// causal order (states: causally-related events have distinct values; concurrent
|
|
// events never collide because the sequence bumps on any wall-clock stall).
|
|
package hlc
|
|
|
|
import (
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// seqBits is the number of low bits reserved for the per-millisecond sequence,
|
|
// giving 2^22 ≈ 4.2M slots per ms — far beyond ingest rates.
|
|
const seqBits = 22
|
|
const seqMask = int64(1)<<seqBits - 1
|
|
const wallShift = seqBits
|
|
|
|
// Clock is a single-writer HLC. It is safe for concurrent use.
|
|
type Clock struct {
|
|
mu sync.Mutex
|
|
wallMS int64 // last observed wall-clock millis
|
|
seq int64 // sequence within the current wallMillis bucket
|
|
}
|
|
|
|
func New() *Clock { return &Clock{} }
|
|
|
|
// SeekTo adopts the given packed HLC value when it is ahead of the clock's current
|
|
// position, so the next Now is still strictly increasing. Used to resume a node's
|
|
// clock from its persisted MAX(hcl) at startup — without it, a restart with a
|
|
// regressed wall clock would reissue already-used values and break the
|
|
// monotonicity the (node_id, hcl) locator uniqueness and gossip cursors rely on.
|
|
func (c *Clock) SeekTo(v int64) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
wall := v >> wallShift
|
|
seq := v & seqMask
|
|
if wall > c.wallMS || (wall == c.wallMS && seq > c.seq) {
|
|
c.wallMS = wall
|
|
c.seq = seq
|
|
}
|
|
}
|
|
|
|
// Now returns the next monotonic HLC value and the wall-clock time embedded in
|
|
// it. The returned time is the HLC's wall component — never ahead of the local
|
|
// clock beyond the current call and never rewinding across calls.
|
|
func (c *Clock) Now() (int64, time.Time) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
w := time.Now().UnixMilli()
|
|
if w > c.wallMS {
|
|
c.wallMS = w
|
|
c.seq = 0
|
|
} else {
|
|
// Wall clock stalled or went backwards (NTP): keep wallMS but bump seq
|
|
// so the value is still strictly increasing.
|
|
if c.seq >= seqMask {
|
|
// Extremely unlikely (4.2M events in one ms); jump the wall lazily.
|
|
c.wallMS++
|
|
c.seq = 0
|
|
} else {
|
|
c.seq++
|
|
}
|
|
}
|
|
|
|
v := c.wallMS<<wallShift | c.seq
|
|
return v, time.UnixMilli(c.wallMS).UTC()
|
|
}
|
|
|
|
// WallTime extracts the wall-clock component embedded in a packed HLC value.
|
|
func WallTime(v int64) time.Time {
|
|
return time.UnixMilli(v >> wallShift).UTC()
|
|
}
|