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
This commit is contained in:
+61
-5
@@ -1,7 +1,9 @@
|
|||||||
package db
|
package db
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"crypto/rand"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
|
"encoding/hex"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -10,6 +12,7 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/david/knox/internal/hlc"
|
||||||
_ "modernc.org/sqlite"
|
_ "modernc.org/sqlite"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -17,6 +20,8 @@ const goldenMinConfidence = 0.7
|
|||||||
|
|
||||||
type KnoxDB struct {
|
type KnoxDB struct {
|
||||||
db *sql.DB
|
db *sql.DB
|
||||||
|
nodeID string
|
||||||
|
clock *hlc.Clock
|
||||||
}
|
}
|
||||||
|
|
||||||
type Observation struct {
|
type Observation struct {
|
||||||
@@ -101,6 +106,9 @@ func Open(path string) (*KnoxDB, error) {
|
|||||||
migrations := []string{
|
migrations := []string{
|
||||||
"ALTER TABLE threads ADD COLUMN provenance TEXT DEFAULT '{}'",
|
"ALTER TABLE threads ADD COLUMN provenance TEXT DEFAULT '{}'",
|
||||||
"ALTER TABLE entries ADD COLUMN created_at 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 {
|
for _, m := range migrations {
|
||||||
if _, err := db.Exec(m); err != nil && !strings.Contains(err.Error(), "duplicate column") {
|
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() }
|
func (k *KnoxDB) Close() error { return k.db.Close() }
|
||||||
|
|
||||||
// RecordObservation appends an immutable observation and updates the entries cache.
|
// 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()
|
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.
|
// Dedup: compare against the latest observation of the same fingerprint.
|
||||||
// For content sources (LineEnd > 0) the extent/summary IS the change signal —
|
// 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(
|
err = tx.QueryRow(
|
||||||
`SELECT id, COALESCE(title,''), COALESCE(summary,''), COALESCE(project,''),
|
`SELECT id, COALESCE(title,''), COALESCE(summary,''), COALESCE(project,''),
|
||||||
COALESCE(created_at,''), COALESCE(line_start,0), COALESCE(line_end,0)
|
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)
|
).Scan(&prevID, &pTitle, &pSummary, &pProject, &pCreated, &pStart, &pEnd)
|
||||||
switch err {
|
switch err {
|
||||||
case nil:
|
case nil:
|
||||||
@@ -154,10 +208,12 @@ func (k *KnoxDB) RecordObservation(o ObservationRecord) (obsID int64, isNew bool
|
|||||||
res, err := tx.Exec(
|
res, err := tx.Exec(
|
||||||
`INSERT INTO observations
|
`INSERT INTO observations
|
||||||
(fingerprint, source_id, source_path, project, content_type, title, summary,
|
(fingerprint, source_id, source_path, project, content_type, title, summary,
|
||||||
collected_at, created_at, line_start, line_end, confidence, ingester_version, trigger, provenance)
|
collected_at, created_at, line_start, line_end, confidence, ingester_version, trigger, provenance,
|
||||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
node_id, hcl, received_at)
|
||||||
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||||
o.Fingerprint, o.SourceID, o.SourcePath, o.Project, o.ContentType, o.Title, o.Summary,
|
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,
|
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 {
|
if err != nil {
|
||||||
return 0, false, fmt.Errorf("insert observation: %w", err)
|
return 0, false, fmt.Errorf("insert observation: %w", err)
|
||||||
|
|||||||
@@ -18,7 +18,10 @@ CREATE TABLE IF NOT EXISTS observations (
|
|||||||
confidence REAL DEFAULT 0.5,
|
confidence REAL DEFAULT 0.5,
|
||||||
ingester_version TEXT DEFAULT '1',
|
ingester_version TEXT DEFAULT '1',
|
||||||
trigger TEXT,
|
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);
|
CREATE INDEX IF NOT EXISTS idx_obs_fingerprint ON observations(fingerprint);
|
||||||
|
|||||||
@@ -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)<<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{} }
|
||||||
|
|
||||||
|
// 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()
|
||||||
|
}
|
||||||
+12
-2
@@ -123,7 +123,12 @@ func (idx *TFIDFIndex) TopTerms(docIdx int, n int) []struct {
|
|||||||
}{t, s})
|
}{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 {
|
if len(scored) > n {
|
||||||
scored = scored[:n]
|
scored = scored[:n]
|
||||||
}
|
}
|
||||||
@@ -220,7 +225,12 @@ func (idx *TFIDFIndex) Cluster(minSharedTerms int, maxTopics int) []TopicCluster
|
|||||||
for k, v := range allTerms {
|
for k, v := range allTerms {
|
||||||
sorted = append(sorted, kv{k, v})
|
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
|
var nameParts []string
|
||||||
for _, kv := range sorted {
|
for _, kv := range sorted {
|
||||||
|
|||||||
+12
-12
@@ -8,7 +8,6 @@ import (
|
|||||||
"regexp"
|
"regexp"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// GitIngester tracks project status for local git repositories.
|
// GitIngester tracks project status for local git repositories.
|
||||||
@@ -141,7 +140,7 @@ type GitStatus struct {
|
|||||||
func (g *GitIngester) repoToResult(repoPath string) *IngestResult {
|
func (g *GitIngester) repoToResult(repoPath string) *IngestResult {
|
||||||
status := gitStatus(repoPath)
|
status := gitStatus(repoPath)
|
||||||
name := filepath.Base(repoPath)
|
name := filepath.Base(repoPath)
|
||||||
fp := Fingerprint([]byte("git:repo:" + repoPath))
|
fp := Fingerprint([]byte("git:repo:" + gitIdentity(repoPath, status.Branch)))
|
||||||
|
|
||||||
clean := "clean"
|
clean := "clean"
|
||||||
if status.Dirty > 0 {
|
if status.Dirty > 0 {
|
||||||
@@ -157,9 +156,6 @@ func (g *GitIngester) repoToResult(repoPath string) *IngestResult {
|
|||||||
}
|
}
|
||||||
|
|
||||||
createdAt := status.LastCommit
|
createdAt := status.LastCommit
|
||||||
if createdAt == "" {
|
|
||||||
createdAt = gitStatusTime(status)
|
|
||||||
}
|
|
||||||
|
|
||||||
return &IngestResult{
|
return &IngestResult{
|
||||||
Fingerprint: fp,
|
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 {
|
func gitStatus(repoPath string) GitStatus {
|
||||||
out, err := gitOut(repoPath, "status", "--porcelain")
|
out, err := gitOut(repoPath, "status", "--porcelain")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -257,6 +246,17 @@ func gitOut(repoPath string, args ...string) (string, error) {
|
|||||||
return sanitizeGitField(string(out)), nil
|
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) {
|
func (g *GitIngester) Ingest(_ string) (*IngestResult, error) {
|
||||||
return nil, fmt.Errorf("use IngestAll() for git")
|
return nil, fmt.Errorf("use IngestAll() for git")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -148,14 +148,16 @@ func (g *GiteaIngester) fetchPulls() ([]teaPull, error) {
|
|||||||
return pulls, nil
|
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 {
|
func signalTime(updated string) string {
|
||||||
if updated != "" {
|
if updated != "" {
|
||||||
if _, err := time.Parse(time.RFC3339, updated); err == nil {
|
if _, err := time.Parse(time.RFC3339, updated); err == nil {
|
||||||
return updated
|
return updated
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return time.Now().UTC().Format(time.RFC3339)
|
return ""
|
||||||
}
|
}
|
||||||
|
|
||||||
func (g *GiteaIngester) repoToResult(r teaRepo) *IngestResult {
|
func (g *GiteaIngester) repoToResult(r teaRepo) *IngestResult {
|
||||||
|
|||||||
@@ -69,7 +69,7 @@ func (l *LogIngester) Ingest(path string) (*IngestResult, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
base := filepath.Base(path)
|
base := filepath.Base(path)
|
||||||
fp := Fingerprint([]byte(path))
|
fp := Fingerprint([]byte("log:" + base))
|
||||||
title := fmt.Sprintf("Log %s (%d lines)", base, totalLines)
|
title := fmt.Sprintf("Log %s (%d lines)", base, totalLines)
|
||||||
createdAt := ""
|
createdAt := ""
|
||||||
if fi, err := os.Stat(path); err == nil {
|
if fi, err := os.Stat(path); err == nil {
|
||||||
@@ -99,7 +99,7 @@ func (l *LogIngester) Ingest(path string) (*IngestResult, error) {
|
|||||||
return &IngestResult{
|
return &IngestResult{
|
||||||
Fingerprint: fp,
|
Fingerprint: fp,
|
||||||
SourceID: l.SourceID(),
|
SourceID: l.SourceID(),
|
||||||
SourcePath: path,
|
SourcePath: base,
|
||||||
ContentType: ".log",
|
ContentType: ".log",
|
||||||
Title: title,
|
Title: title,
|
||||||
Summary: summary,
|
Summary: summary,
|
||||||
|
|||||||
@@ -84,13 +84,14 @@ func (o *ObsidianIngester) ingestNote(path, vault string) (*IngestResult, error)
|
|||||||
modified = created
|
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{
|
fp := FingerprintWithMeta(data, map[string]string{
|
||||||
"source": "obsidian",
|
"source": "obsidian",
|
||||||
"path": path,
|
"path": relPath,
|
||||||
})
|
})
|
||||||
|
|
||||||
relPath, _ := filepath.Rel(vault, path)
|
|
||||||
summary := extractBodyPreview(content, 200)
|
summary := extractBodyPreview(content, 200)
|
||||||
|
|
||||||
return &IngestResult{
|
return &IngestResult{
|
||||||
|
|||||||
@@ -40,12 +40,12 @@ func (o *ObsidianFileIngester) Ingest(path string) (*IngestResult, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
relPath, _ := filepath.Rel(o.Vault, path)
|
||||||
fp := FingerprintWithMeta(data, map[string]string{
|
fp := FingerprintWithMeta(data, map[string]string{
|
||||||
"source": "obsidian",
|
"source": "obsidian",
|
||||||
"path": path,
|
"path": relPath,
|
||||||
})
|
})
|
||||||
|
|
||||||
relPath, _ := filepath.Rel(o.Vault, path)
|
|
||||||
summary := extractBodyPreview(content, 200)
|
summary := extractBodyPreview(content, 200)
|
||||||
|
|
||||||
return &IngestResult{
|
return &IngestResult{
|
||||||
|
|||||||
Reference in New Issue
Block a user