From 2c0f8fa257d22a3e069f5b89cf9894e5ef423a11 Mon Sep 17 00:00:00 2001 From: David Gwilliam Date: Sat, 29 Aug 2026 03:51:02 -0700 Subject: [PATCH] feat: M1 determinism fixes for gossip Refs #1 - F1: canonical fingerprints (git identity by remote URL, log by basename, obsidian by relative vault path) so identical facts get identical ids across machines - F2: node_id (persisted in settings) + Hybrid Logical Clock in all observation "when" columns; removed time.Now() fact-time fallbacks - F3: stable TF-IDF tie-break sort (score desc, term asc) - F4: observations carry (node_id, hcl) locator; dedup ordered by hcl --- internal/db/db.go | 68 +++++++++++++++++++++++++++++--- internal/db/schema.go | 5 ++- internal/hlc/hlc.go | 60 ++++++++++++++++++++++++++++ internal/index/tfidf.go | 14 ++++++- internal/ingest/git.go | 24 +++++------ internal/ingest/gitea.go | 6 ++- internal/ingest/log.go | 4 +- internal/ingest/obsidian.go | 7 ++-- internal/ingest/obsidian_file.go | 4 +- 9 files changed, 162 insertions(+), 30 deletions(-) create mode 100644 internal/hlc/hlc.go diff --git a/internal/db/db.go b/internal/db/db.go index f9211b9..1b12d26 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -1,7 +1,9 @@ package db import ( + "crypto/rand" "database/sql" + "encoding/hex" "fmt" "os" "path/filepath" @@ -10,13 +12,16 @@ import ( "sync" "time" + "github.com/david/knox/internal/hlc" _ "modernc.org/sqlite" ) const goldenMinConfidence = 0.7 type KnoxDB struct { - db *sql.DB + db *sql.DB + nodeID string + clock *hlc.Clock } type Observation struct { @@ -101,6 +106,9 @@ func Open(path string) (*KnoxDB, error) { migrations := []string{ "ALTER TABLE threads ADD COLUMN provenance TEXT DEFAULT '{}'", "ALTER TABLE entries ADD COLUMN created_at TEXT DEFAULT ''", + "ALTER TABLE observations ADD COLUMN node_id TEXT NOT NULL DEFAULT ''", + "ALTER TABLE observations ADD COLUMN hcl INTEGER", + "ALTER TABLE observations ADD COLUMN received_at TEXT", } for _, m := range migrations { if _, err := db.Exec(m); err != nil && !strings.Contains(err.Error(), "duplicate column") { @@ -108,9 +116,54 @@ func Open(path string) (*KnoxDB, error) { } } - return &KnoxDB{db: db}, nil + kdb := &KnoxDB{db: db, clock: hlc.New()} + + // Node identity: generated once, persisted in settings. Stable across + // restarts so gossip cursors and locator IDs survive. + nodeID, err := kdb.loadNodeID() + if err != nil { + return nil, err + } + kdb.nodeID = nodeID + + // Backfill: rows added before node_id existed belong to this node. + if _, err := db.Exec("UPDATE observations SET node_id=? WHERE node_id=''", nodeID); err != nil { + return nil, fmt.Errorf("backfill node_id: %w", err) + } + + return kdb, nil } +const nodeIDKey = "node_id" + +// loadNodeID reads the persisted node identity, creating one if absent. +func (k *KnoxDB) loadNodeID() (string, error) { + var id string + err := k.db.QueryRow(`SELECT value FROM settings WHERE key=?`, nodeIDKey).Scan(&id) + if err == nil && id != "" { + return id, nil + } + if err != nil && err != sql.ErrNoRows { + return "", fmt.Errorf("read node_id: %w", err) + } + + buf := make([]byte, 16) + if _, err := rand.Read(buf); err != nil { + return "", fmt.Errorf("generate node_id: %w", err) + } + id = hex.EncodeToString(buf) + if _, err := k.db.Exec( + `INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value=?`, + nodeIDKey, id, id, + ); err != nil { + return "", fmt.Errorf("persist node_id: %w", err) + } + return id, nil +} + +// NodeID returns this node's stable identity. +func (k *KnoxDB) NodeID() string { return k.nodeID } + func (k *KnoxDB) Close() error { return k.db.Close() } // RecordObservation appends an immutable observation and updates the entries cache. @@ -123,7 +176,8 @@ func (k *KnoxDB) RecordObservation(o ObservationRecord) (obsID int64, isNew bool } defer tx.Rollback() - now := time.Now().UTC().Format(time.RFC3339) + hclVal, hclWall := k.clock.Now() + now := hclWall.Format(time.RFC3339) // Dedup: compare against the latest observation of the same fingerprint. // For content sources (LineEnd > 0) the extent/summary IS the change signal — @@ -136,7 +190,7 @@ func (k *KnoxDB) RecordObservation(o ObservationRecord) (obsID int64, isNew bool err = tx.QueryRow( `SELECT id, COALESCE(title,''), COALESCE(summary,''), COALESCE(project,''), COALESCE(created_at,''), COALESCE(line_start,0), COALESCE(line_end,0) - FROM observations WHERE fingerprint=? ORDER BY id DESC LIMIT 1`, o.Fingerprint, + FROM observations WHERE fingerprint=? ORDER BY hcl DESC, id DESC LIMIT 1`, o.Fingerprint, ).Scan(&prevID, &pTitle, &pSummary, &pProject, &pCreated, &pStart, &pEnd) switch err { case nil: @@ -154,10 +208,12 @@ func (k *KnoxDB) RecordObservation(o ObservationRecord) (obsID int64, isNew bool res, err := tx.Exec( `INSERT INTO observations (fingerprint, source_id, source_path, project, content_type, title, summary, - collected_at, created_at, line_start, line_end, confidence, ingester_version, trigger, provenance) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + collected_at, created_at, line_start, line_end, confidence, ingester_version, trigger, provenance, + node_id, hcl, received_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, o.Fingerprint, o.SourceID, o.SourcePath, o.Project, o.ContentType, o.Title, o.Summary, now, o.CreatedAt, o.LineStart, o.LineEnd, o.Confidence, o.IngesterVersion, o.Trigger, o.Provenance, + k.nodeID, hclVal, time.Now().UTC().Format(time.RFC3339), ) if err != nil { return 0, false, fmt.Errorf("insert observation: %w", err) diff --git a/internal/db/schema.go b/internal/db/schema.go index 06b1c7d..6a37d02 100644 --- a/internal/db/schema.go +++ b/internal/db/schema.go @@ -18,7 +18,10 @@ CREATE TABLE IF NOT EXISTS observations ( confidence REAL DEFAULT 0.5, ingester_version TEXT DEFAULT '1', trigger TEXT, - provenance TEXT DEFAULT '{}' + provenance TEXT DEFAULT '{}', + node_id TEXT NOT NULL DEFAULT '', + hcl INTEGER, + received_at TEXT ); CREATE INDEX IF NOT EXISTS idx_obs_fingerprint ON observations(fingerprint); diff --git a/internal/hlc/hlc.go b/internal/hlc/hlc.go new file mode 100644 index 0000000..2827e16 --- /dev/null +++ b/internal/hlc/hlc.go @@ -0,0 +1,60 @@ +// 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)< 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).UTC() +} \ No newline at end of file diff --git a/internal/index/tfidf.go b/internal/index/tfidf.go index 85c2482..81c7987 100644 --- a/internal/index/tfidf.go +++ b/internal/index/tfidf.go @@ -123,7 +123,12 @@ func (idx *TFIDFIndex) TopTerms(docIdx int, n int) []struct { }{t, s}) } } - sort.Slice(scored, func(i, j int) bool { return scored[i].Score > scored[j].Score }) + sort.Slice(scored, func(i, j int) bool { + if scored[i].Score != scored[j].Score { + return scored[i].Score > scored[j].Score + } + return scored[i].Term < scored[j].Term + }) if len(scored) > n { scored = scored[:n] } @@ -220,7 +225,12 @@ func (idx *TFIDFIndex) Cluster(minSharedTerms int, maxTopics int) []TopicCluster for k, v := range allTerms { sorted = append(sorted, kv{k, v}) } - sort.Slice(sorted, func(i, j int) bool { return sorted[i].v > sorted[j].v }) + sort.Slice(sorted, func(i, j int) bool { + if sorted[i].v != sorted[j].v { + return sorted[i].v > sorted[j].v + } + return sorted[i].k < sorted[j].k + }) var nameParts []string for _, kv := range sorted { diff --git a/internal/ingest/git.go b/internal/ingest/git.go index e9eb5a0..ae0e659 100644 --- a/internal/ingest/git.go +++ b/internal/ingest/git.go @@ -8,7 +8,6 @@ import ( "regexp" "sort" "strings" - "time" ) // GitIngester tracks project status for local git repositories. @@ -141,7 +140,7 @@ type GitStatus struct { func (g *GitIngester) repoToResult(repoPath string) *IngestResult { status := gitStatus(repoPath) name := filepath.Base(repoPath) - fp := Fingerprint([]byte("git:repo:" + repoPath)) + fp := Fingerprint([]byte("git:repo:" + gitIdentity(repoPath, status.Branch))) clean := "clean" if status.Dirty > 0 { @@ -157,9 +156,6 @@ func (g *GitIngester) repoToResult(repoPath string) *IngestResult { } createdAt := status.LastCommit - if createdAt == "" { - createdAt = gitStatusTime(status) - } return &IngestResult{ Fingerprint: fp, @@ -186,13 +182,6 @@ func (g *GitIngester) repoToResult(repoPath string) *IngestResult { } } -func gitStatusTime(s GitStatus) string { - if s.LastCommit != "" { - return s.LastCommit - } - return time.Now().UTC().Format(time.RFC3339) -} - func gitStatus(repoPath string) GitStatus { out, err := gitOut(repoPath, "status", "--porcelain") if err != nil { @@ -257,6 +246,17 @@ func gitOut(repoPath string, args ...string) (string, error) { return sanitizeGitField(string(out)), nil } +// gitIdentity returns a machine-independent identity for a repo: the remote +// origin URL when configured (works across clones/machines), else the repo +// directory name. The branch is appended so entries are per-branch like the +// underlying gitStatus. +func gitIdentity(repoPath, branch string) string { + if out, err := gitOut(repoPath, "remote", "get-url", "origin"); err == nil && out != "" { + return out + ":" + branch + } + return filepath.Base(repoPath) + ":" + branch +} + func (g *GitIngester) Ingest(_ string) (*IngestResult, error) { return nil, fmt.Errorf("use IngestAll() for git") } diff --git a/internal/ingest/gitea.go b/internal/ingest/gitea.go index 050e113..c5e2b75 100644 --- a/internal/ingest/gitea.go +++ b/internal/ingest/gitea.go @@ -148,14 +148,16 @@ func (g *GiteaIngester) fetchPulls() ([]teaPull, error) { return pulls, nil } -// signalTime returns the item's last-activity time, falling back to now. +// signalTime returns the item's last-activity time, or "" when unknown. An +// empty value is an explicit unknown — downstream uses it as a missing signal, +// never fabricates a "now" timestamp that would differ across nodes. func signalTime(updated string) string { if updated != "" { if _, err := time.Parse(time.RFC3339, updated); err == nil { return updated } } - return time.Now().UTC().Format(time.RFC3339) + return "" } func (g *GiteaIngester) repoToResult(r teaRepo) *IngestResult { diff --git a/internal/ingest/log.go b/internal/ingest/log.go index c564550..5f6033d 100644 --- a/internal/ingest/log.go +++ b/internal/ingest/log.go @@ -69,7 +69,7 @@ func (l *LogIngester) Ingest(path string) (*IngestResult, error) { } base := filepath.Base(path) - fp := Fingerprint([]byte(path)) + fp := Fingerprint([]byte("log:" + base)) title := fmt.Sprintf("Log %s (%d lines)", base, totalLines) createdAt := "" if fi, err := os.Stat(path); err == nil { @@ -99,7 +99,7 @@ func (l *LogIngester) Ingest(path string) (*IngestResult, error) { return &IngestResult{ Fingerprint: fp, SourceID: l.SourceID(), - SourcePath: path, + SourcePath: base, ContentType: ".log", Title: title, Summary: summary, diff --git a/internal/ingest/obsidian.go b/internal/ingest/obsidian.go index 99d6549..8a19040 100644 --- a/internal/ingest/obsidian.go +++ b/internal/ingest/obsidian.go @@ -84,13 +84,14 @@ func (o *ObsidianIngester) ingestNote(path, vault string) (*IngestResult, error) modified = created } - // Fingerprint by content (so edits create updates) + // Fingerprint by content + relative vault path (machine-independent): + // the same note on two hosts maps to the same fingerprint. + relPath, _ := filepath.Rel(vault, path) fp := FingerprintWithMeta(data, map[string]string{ "source": "obsidian", - "path": path, + "path": relPath, }) - relPath, _ := filepath.Rel(vault, path) summary := extractBodyPreview(content, 200) return &IngestResult{ diff --git a/internal/ingest/obsidian_file.go b/internal/ingest/obsidian_file.go index e64c201..ea81323 100644 --- a/internal/ingest/obsidian_file.go +++ b/internal/ingest/obsidian_file.go @@ -40,12 +40,12 @@ func (o *ObsidianFileIngester) Ingest(path string) (*IngestResult, error) { } } + relPath, _ := filepath.Rel(o.Vault, path) fp := FingerprintWithMeta(data, map[string]string{ "source": "obsidian", - "path": path, + "path": relPath, }) - relPath, _ := filepath.Rel(o.Vault, path) summary := extractBodyPreview(content, 200) return &IngestResult{