diff --git a/CHANGELOG.md b/CHANGELOG.md index f41bd8a..c15e9dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,76 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed +- **Standalone module resolution**: `go.mod` required + `github.com/GrayCodeAI/hawk-mcpkit v0.0.0` (a non-existent version, only + satisfiable through the monorepo `replace`), which made yaad unbuildable + whenever it was consumed as a dependency module. The requirement is now the + tagged `v0.1.5` (which ships `ServeHTTPWithShutdown`, the newest mcpkit API + yaad uses) and `go.sum` carries its hashes; the local `replace` remains for + monorepo development. The dead `ServeSSE`/`ServeSSEWithShutdown` wrappers in + `internal/server` were removed — they referenced mcpkit APIs added after + v0.1.5 and had no callers in yaad. +- **Per-connection PRAGMAs now cover the whole connection pool**: + `synchronous`, `wal_autocheckpoint`, `temp_store`, `mmap_size`, + `cache_size`, and `recursive_triggers` were issued once through + `db.ExecContext`, which only configured the single pooled connection that + served the call — the other four connections (pool size 5) silently ran on + SQLite defaults. They are now `_pragma=` DSN parameters, which the + modernc.org/sqlite driver applies to every new connection. `page_size` + (a no-op after the database file exists) was dropped, and `PRAGMA optimize` + moved from startup to `Store.Close()`, matching SQLite's guidance to run it + near connection close. +- **Batch metadata and edge-count queries chunk large ID sets**: + `LoadNodeMetadata`/`FillNodeMetadata` and `CountEdgesBatch` built a single + `IN (...)` with every requested ID, ignoring the `maxSQLVariables` (900) + host-parameter budget the other batch helpers (`GetNodesBatch`, + `GetAllEdgesFor`, `GetEdgesBetween`) already chunk at — large ID sets could + exceed SQLite's per-statement parameter limit. Both now fetch in + `maxSQLVariables`-sized chunks and merge results, with regression tests + covering >900 IDs. +- **Backup failures are now visible and backups are fsynced**: + - `BackupScheduler.run` discarded every snapshot error (`_ = b.RunNow`), so + a dead backup pipeline looked healthy. Failures are now logged via + `log/slog` and tracked in the new `BackupScheduler.Status()` accessor + (last success/error timestamps, snapshot/failure counters), with + `RunNow` recording its outcome too. + - `RotateBackups` silently ignored `os.Remove` errors; unexpected ones are + now logged (a vanished file is still fine). + - `Store.Backup` now fsyncs the snapshot file before renaming it into + place and best-effort fsyncs the backup directory after the rename, so a + crash cannot leave a renamed-but-not-durable (or truncated) backup. +- **`EnvKeyProvider` enforces a minimum key policy**: the raw environment + value was used as master key material regardless of length, so a short + passphrase (whose entropy HKDF cannot fix) was silently accepted. + `KeyBytes` now accepts a raw string of at least 32 bytes, or a base64- or + hex-encoded value that decodes to exactly 32 bytes (raw wins when a value + qualifies both ways), and otherwise returns an error naming the + environment variable. + **Upgrade note:** deployments using a shorter key are rejected after this + change (in hawk this disables the yaad memory bridge with a logged + warning — it never falls back to plaintext). Rows already encrypted under + a now-rejected short key cannot be decrypted by `EnvKeyProvider` anymore; + set a compliant key and re-create or migrate that data. +- **Process lock paths are normalized**: `AcquireProcessLock` keys its + in-process lock registry by the raw `dbPath + ".lock"` string, so + lexically equivalent paths (`dir/db`, `dir/./db`, `dir/x/../db`) opened two + lock-file handles in one process and the second open falsely reported + "database locked by another yaad process". Lock paths are now normalized + with `filepath.Abs`. +- **`storage.DeleteNode` foreign-key violations**: `deleteNodeQ` only deleted + `edges` + `nodes`, so deleting a node with child rows in `embeddings`, + `embeddings_hnsw`, `file_watch`, or `node_signatures` failed with + `FOREIGN KEY constraint failed` (no `ON DELETE CASCADE` in the schema), and + `node_versions` / `node_metadata` rows leaked as orphans. All child rows are + now deleted explicitly, leaf tables first, in the same transaction. +- **Swallowed delete errors in engine passes**: `GarbageCollect` + (`engine/decay.go`), the sparsifier passes (`engine/sparsify.go`), + `consolidateDuplicates` (`engine/improve.go`), and the LLM consolidator + (`engine/llm_consolidation.go`) silently ignored `DeleteNode` failures and + mis-reported removed/merged/pruned counts. Failures are now logged via + `log/slog` and only successful deletions are counted. + ## [0.2.0] — 2026-07-14 ### Changed diff --git a/engine/decay.go b/engine/decay.go index d4e32ed..7f116db 100644 --- a/engine/decay.go +++ b/engine/decay.go @@ -2,6 +2,7 @@ package engine import ( "context" + "log/slog" "math" "time" @@ -142,13 +143,20 @@ func GarbageCollect(ctx context.Context, store storage.Storage, cfg DecayConfig) // Phase 2: delete all collected IDs removed := 0 + failed := 0 for _, id := range toDelete { if err := ctx.Err(); err != nil { return removed, err } - if err := store.DeleteNode(ctx, id); err == nil { - removed++ + if err := store.DeleteNode(ctx, id); err != nil { + failed++ + slog.Warn("gc: delete node failed", "node_id", id, "error", err) + continue } + removed++ + } + if failed > 0 { + slog.Error("gc: failed to delete nodes", "failed", failed, "removed", removed) } return removed, nil } diff --git a/engine/improve.go b/engine/improve.go index 61365b9..a66278a 100644 --- a/engine/improve.go +++ b/engine/improve.go @@ -16,6 +16,7 @@ package engine import ( "context" "fmt" + "log/slog" "strings" "sync/atomic" "time" @@ -241,6 +242,7 @@ func (e *Engine) consolidateDuplicates(ctx context.Context, nodes []*storage.Nod continue } if err := e.store.DeleteNode(ctx, n.ID); err != nil { + slog.Warn("improve: failed to delete duplicate node", "node_id", n.ID, "error", err) continue } merged++ diff --git a/engine/llm_consolidation.go b/engine/llm_consolidation.go index d35591f..8b5ab8a 100644 --- a/engine/llm_consolidation.go +++ b/engine/llm_consolidation.go @@ -3,6 +3,7 @@ package engine import ( "context" "fmt" + "log/slog" "sort" "strings" "unicode" @@ -249,7 +250,7 @@ func (lc *LLMConsolidator) applyMergePlan(ctx context.Context, plan *MergePlan) for _, nid := range plan.NodeIDs { if err := lc.store.DeleteNode(ctx, nid); err != nil { // Best-effort delete; log but don't fail. - continue + slog.Warn("llm-consolidation: failed to delete merged-away node", "node_id", nid, "error", err) } } diff --git a/engine/sparsify.go b/engine/sparsify.go index 9633356..2459749 100644 --- a/engine/sparsify.go +++ b/engine/sparsify.go @@ -3,6 +3,7 @@ package engine import ( "context" "fmt" + "log/slog" "sort" "strings" @@ -128,7 +129,9 @@ func (s *Sparsifier) mergeNearDuplicates(ctx context.Context) (int, error) { _ = s.store.UpdateNodeContent(ctx, primary.ID, combined) } // Archive the duplicate. - _ = s.store.DeleteNode(ctx, duplicate.ID) + if err := s.store.DeleteNode(ctx, duplicate.ID); err != nil { + slog.Warn("sparsify: failed to delete merged duplicate", "node_id", duplicate.ID, "error", err) + } processed[duplicate.ID] = true merged++ } @@ -200,7 +203,10 @@ func (s *Sparsifier) compressLowValueClusters(ctx context.Context) (int, error) // Delete compressed nodes for _, n := range toCompress { - _ = s.store.DeleteNode(ctx, n.ID) + if err := s.store.DeleteNode(ctx, n.ID); err != nil { + slog.Warn("sparsify: failed to delete compressed node", "node_id", n.ID, "error", err) + continue + } compressed++ } } @@ -226,7 +232,10 @@ func (s *Sparsifier) pruneOrphans(ctx context.Context) (int, error) { continue } if inbound+outbound == 0 && n.AccessCount <= 1 { - _ = s.store.DeleteNode(ctx, n.ID) + if err := s.store.DeleteNode(ctx, n.ID); err != nil { + slog.Warn("sparsify: failed to prune orphan node", "node_id", n.ID, "error", err) + continue + } pruned++ } } diff --git a/go.mod b/go.mod index 3ee17fe..284d6be 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.26.6 require ( github.com/BurntSushi/toml v1.6.0 github.com/GrayCodeAI/hawk-core-contracts v0.1.8 - github.com/GrayCodeAI/hawk-mcpkit v0.0.0 + github.com/GrayCodeAI/hawk-mcpkit v0.1.5 github.com/google/uuid v1.6.0 github.com/mark3labs/mcp-go v0.49.0 github.com/prometheus/client_golang v1.24.1 @@ -48,9 +48,9 @@ require ( modernc.org/memory v1.11.0 // indirect ) -// replace: use local hawk-mcpkit during monorepo development. -// yaad depends on ServeHTTPWithShutdown and other APIs added after v0.1.4; -// switch to a tagged version once v0.2.0 is released. +// replace: use local hawk-mcpkit during monorepo development. The require +// above pins the minimum tagged release (v0.1.5 ships ServeHTTPWithShutdown, +// which yaad needs); drop this replace when building yaad standalone. replace github.com/GrayCodeAI/hawk-mcpkit => ../hawk-mcpkit replace github.com/GrayCodeAI/hawk-core-contracts => ../hawk-core-contracts diff --git a/internal/server/mcp_core.go b/internal/server/mcp_core.go index eb953e0..72b790b 100644 --- a/internal/server/mcp_core.go +++ b/internal/server/mcp_core.go @@ -56,19 +56,6 @@ func (s *MCPServer) ServeHTTPWithShutdown(addr string) (*mcpserver.StreamableHTT return s.srv.ServeHTTPWithShutdown(addr) } -// ServeSSE starts the MCP server on the SSE (Server-Sent Events) transport. -// This is the classic MCP transport used by Claude Desktop. -func (s *MCPServer) ServeSSE(addr string) error { - return s.srv.ServeSSE(addr) -} - -// ServeSSEWithShutdown starts the MCP server on the SSE transport and -// returns the underlying server so the caller can invoke Shutdown for -// graceful teardown. -func (s *MCPServer) ServeSSEWithShutdown(addr string) (*mcpserver.SSEServer, error) { - return s.srv.ServeSSEWithShutdown(addr) -} - // mcp returns the underlying mcp-go server so the progress notification // helper can reach mcp-go's send path directly. func (s *MCPServer) mcp() *mcpserver.MCPServer { diff --git a/storage/backup.go b/storage/backup.go index c79a386..5119fdc 100644 --- a/storage/backup.go +++ b/storage/backup.go @@ -37,9 +37,40 @@ func (s *Store) Backup(ctx context.Context, backupPath string) error { _ = os.Remove(tmp) return fmt.Errorf("restrict backup file permissions: %w", err) } + // fsync the snapshot before the rename: once renamed, the file reads as + // a complete backup, so its contents must already be on durable storage. + if err := syncFile(tmp); err != nil { + _ = os.Remove(tmp) + return fmt.Errorf("sync backup file: %w", err) + } if err := os.Rename(tmp, backupPath); err != nil { _ = os.Remove(tmp) return fmt.Errorf("finalize backup: %w", err) } + // Best-effort fsync of the directory so the rename itself survives a + // crash. Directory Sync is a no-op / unsupported on some platforms + // (Windows rejects fsync on directory handles), hence the ignore. + _ = syncDir(dir) return nil } + +// syncFile flushes a file's contents to durable storage. +func syncFile(path string) error { + f, err := os.Open(path) + if err != nil { + return err + } + defer func() { _ = f.Close() }() + return f.Sync() +} + +// syncDir fsyncs a directory entry so a rename inside it is durable. Errors +// are ignored by callers: several platforms cannot sync directory handles. +func syncDir(path string) error { + d, err := os.Open(path) + if err != nil { + return err + } + defer func() { _ = d.Close() }() + return d.Sync() +} diff --git a/storage/backup_rotate.go b/storage/backup_rotate.go index d84e947..f2b8330 100644 --- a/storage/backup_rotate.go +++ b/storage/backup_rotate.go @@ -2,7 +2,10 @@ package storage import ( "context" + "errors" "fmt" + "io/fs" + "log/slog" "os" "path/filepath" "sort" @@ -10,6 +13,15 @@ import ( "time" ) +// removeBackupFile deletes a pruned backup (or stale temp file), logging any +// error other than "already gone" so rotation failures are visible instead of +// silently leaving the directory over retention. +func removeBackupFile(path string) { + if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) { + slog.Warn("backup rotation: failed to remove backup file", "path", path, "error", err) + } +} + // defaultTmpMaxAge bounds how long aborted Backup temp files may linger // before RotateBackups sweeps them. const defaultTmpMaxAge = 24 * time.Hour @@ -49,7 +61,7 @@ func (s *Store) RotateBackups(ctx context.Context, dir string, keep int, maxAge // Sweep abandoned VACUUM INTO temp files from failed backups. if strings.HasSuffix(name, ".tmp") { if time.Since(info.ModTime()) > defaultTmpMaxAge { - _ = os.Remove(path) + removeBackupFile(path) } continue } @@ -70,7 +82,7 @@ func (s *Store) RotateBackups(ctx context.Context, dir string, keep int, maxAge tooMany := keep > 0 && i >= keep tooOld := maxAge > 0 && time.Since(b.modTime) > maxAge && i > 0 if tooMany || tooOld { - _ = os.Remove(b.path) + removeBackupFile(b.path) } } return nil diff --git a/storage/backup_scheduler.go b/storage/backup_scheduler.go index 2b5f4f9..94b8a88 100644 --- a/storage/backup_scheduler.go +++ b/storage/backup_scheduler.go @@ -3,6 +3,7 @@ package storage import ( "context" "fmt" + "log/slog" "os" "path/filepath" "sync" @@ -26,6 +27,52 @@ type BackupScheduler struct { stopped bool stop chan struct{} stoppedW chan struct{} // closed when the run loop returns + + statusMu sync.Mutex // guards the fields below + lastSuccessAt time.Time + lastError error + lastErrorAt time.Time + snapshotCount int64 + snapshotFails int64 +} + +// BackupStatus reports the outcome of the scheduler's snapshots so callers +// (and health checks) can tell whether backups are actually succeeding. +type BackupStatus struct { + LastSuccessAt time.Time // zero until the first successful snapshot + LastError error // nil until the first failed snapshot + LastErrorAt time.Time // zero until the first failed snapshot + Snapshots int64 // successful snapshot count + Failures int64 // failed snapshot count +} + +// Status returns the scheduler's snapshot history. Safe to call from any +// goroutine; also valid before Start (all counters zero). +func (b *BackupScheduler) Status() BackupStatus { + b.statusMu.Lock() + defer b.statusMu.Unlock() + return BackupStatus{ + LastSuccessAt: b.lastSuccessAt, + LastError: b.lastError, + LastErrorAt: b.lastErrorAt, + Snapshots: b.snapshotCount, + Failures: b.snapshotFails, + } +} + +// recordResult updates the scheduler status for one snapshot attempt. +func (b *BackupScheduler) recordResult(err error) { + b.statusMu.Lock() + defer b.statusMu.Unlock() + if err != nil { + b.lastError = err + b.lastErrorAt = time.Now().UTC() + b.snapshotFails++ + return + } + b.lastError = nil + b.lastSuccessAt = time.Now().UTC() + b.snapshotCount++ } // ScheduleBackups registers a scheduler that snapshots the store into dir @@ -95,26 +142,34 @@ func (b *BackupScheduler) Stop() { // RunNow takes an immediate snapshot and rotates the directory. Names are // UTC-timestamp based with a nano-second suffix so snapshots taken within // the same second never collide (VACUUM INTO refuses existing targets). +// The outcome is recorded in Status(). func (b *BackupScheduler) RunNow(ctx context.Context) error { now := time.Now().UTC() name := filepath.Join(b.dir, fmt.Sprintf("yaad-%s-%d.db", now.Format("20060102T150405Z"), now.UnixNano())) if err := b.store.Backup(ctx, name); err != nil { + b.recordResult(err) + return err + } + if err := b.store.RotateBackups(ctx, b.dir, b.keep, b.maxAge); err != nil { + b.recordResult(err) return err } - return b.store.RotateBackups(ctx, b.dir, b.keep, b.maxAge) + b.recordResult(nil) + return nil } // run takes an initial snapshot, then ticks every interval until Stop is // called. Each snapshot is best-effort: a failed attempt is skipped rather -// than retry-looping; the next tick tries again. +// than retry-looping; the next tick tries again. Failures are logged and +// recorded in Status() so silent backup death is observable. func (b *BackupScheduler) run() { defer close(b.stoppedW) // Snapshot once immediately so a short-lived host still leaves a // backup behind; rotation prunes duplicates from frequent restarts. ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) - _ = b.RunNow(ctx) + b.snapshot(ctx) cancel() ticker := time.NewTicker(b.interval) @@ -126,8 +181,16 @@ func (b *BackupScheduler) run() { return case <-ticker.C: ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) - _ = b.RunNow(ctx) + b.snapshot(ctx) cancel() } } } + +// snapshot runs one RunNow attempt, logging failures; RunNow records the +// outcome either way. +func (b *BackupScheduler) snapshot(ctx context.Context) { + if err := b.RunNow(ctx); err != nil { + slog.Warn("backup scheduler: snapshot failed", "dir", b.dir, "error", err) + } +} diff --git a/storage/backup_scheduler_test.go b/storage/backup_scheduler_test.go index 7dfbe0a..e2302c0 100644 --- a/storage/backup_scheduler_test.go +++ b/storage/backup_scheduler_test.go @@ -133,3 +133,51 @@ func TestScheduleBackupsEmptyDir(t *testing.T) { t.Fatal("expected error for empty backup directory") } } + +// TestSchedulerStatusTracksFailures verifies the scheduler surfaces snapshot +// failures instead of discarding them: a successful RunNow updates +// LastSuccessAt/Snapshots, and a failing RunNow records LastError and bumps +// Failures. +func TestSchedulerStatusTracksFailures(t *testing.T) { + s, cleanup := setupStore(t) + defer cleanup() + ctx := context.Background() + + dir := filepath.Join(t.TempDir(), "backups") + sched, err := s.ScheduleBackups(dir, time.Hour, 0, 0) + if err != nil { + t.Fatalf("ScheduleBackups: %v", err) + } + + if st := sched.Status(); st.Snapshots != 0 || st.Failures != 0 || st.LastError != nil || !st.LastSuccessAt.IsZero() { + t.Fatalf("fresh scheduler status = %+v, want zero values", st) + } + + if err := sched.RunNow(ctx); err != nil { + t.Fatalf("RunNow: %v", err) + } + st := sched.Status() + if st.Snapshots != 1 || st.Failures != 0 || st.LastError != nil { + t.Errorf("after success: %+v, want Snapshots=1 Failures=0 LastError=nil", st) + } + if st.LastSuccessAt.IsZero() { + t.Error("LastSuccessAt not set after successful snapshot") + } + + // Make the directory unwritable so the snapshot fails. + if err := os.Chmod(dir, 0o500); err != nil { + t.Fatalf("chmod dir read-only: %v", err) + } + defer func() { _ = os.Chmod(dir, 0o700) }() + + if err := sched.RunNow(ctx); err == nil { + t.Fatal("expected RunNow to fail with unwritable backup dir") + } + st = sched.Status() + if st.Failures != 1 || st.LastError == nil || st.LastErrorAt.IsZero() { + t.Errorf("after failure: %+v, want Failures=1 LastError set LastErrorAt set", st) + } + if st.Snapshots != 1 { + t.Errorf("Snapshots = %d, want still 1", st.Snapshots) + } +} diff --git a/storage/crypto.go b/storage/crypto.go index 1242814..d6dd25c 100644 --- a/storage/crypto.go +++ b/storage/crypto.go @@ -6,6 +6,7 @@ import ( "crypto/rand" "crypto/sha256" "encoding/base64" + "encoding/hex" "fmt" "io" "os" @@ -65,6 +66,17 @@ func NewEnvKeyProvider(envVar string) *EnvKeyProvider { func (p *EnvKeyProvider) CurrentVersion() int { return 0 } // KeyBytes implements KeyProvider. +// +// Accepted material (a raw string wins when the value qualifies both ways): +// - a raw string of at least 32 bytes, used directly as key material; or +// - a base64- or hex-encoded string that decodes to exactly 32 bytes +// (e.g. `openssl rand -base64 32` / `openssl rand -hex 32` output), +// decoded and used as-is. +// +// Anything else is rejected: HKDF would happily stretch a short password +// into an AES-256 key, but the entropy would still be that of the password, +// inviting brute-force attacks directly on the ciphertext. The error names +// the environment variable so operators can fix the configuration. func (p *EnvKeyProvider) KeyBytes(version int) ([]byte, error) { if version != 0 { return nil, fmt.Errorf("%w: %d", ErrUnknownKeyVersion, version) @@ -73,7 +85,24 @@ func (p *EnvKeyProvider) KeyBytes(version int) ([]byte, error) { if raw == "" { return nil, fmt.Errorf("%w: %s is unset or empty", ErrKeyMaterial, p.envVar) } - return []byte(raw), nil + if len(raw) >= 32 { + return []byte(raw), nil + } + for _, dec := range []func(string) ([]byte, error){ + base64.StdEncoding.DecodeString, + base64.RawStdEncoding.DecodeString, + base64.URLEncoding.DecodeString, + base64.RawURLEncoding.DecodeString, + hex.DecodeString, + } { + if b, err := dec(raw); err == nil && len(b) == 32 { + return b, nil + } + } + return nil, fmt.Errorf( + "%w: %s must hold a raw string of at least 32 bytes, or base64/hex text encoding exactly 32 bytes (got %d characters)", + ErrKeyMaterial, p.envVar, len(raw), + ) } // NodeCipher encrypts and decrypts single string values with diff --git a/storage/crypto_test.go b/storage/crypto_test.go index d460dc9..e51312c 100644 --- a/storage/crypto_test.go +++ b/storage/crypto_test.go @@ -1,8 +1,11 @@ package storage import ( + "bytes" "context" "database/sql" + "encoding/base64" + "encoding/hex" "errors" "fmt" "strings" @@ -130,13 +133,13 @@ func TestNewNodeCipherEagerValidation(t *testing.T) { } func TestEnvKeyProvider(t *testing.T) { - t.Setenv("YAAD_TEST_KEY", "env-provided-key-material") + t.Setenv("YAAD_TEST_KEY", "env-provided-key-material-at-least-32-bytes!") p := NewEnvKeyProvider("YAAD_TEST_KEY") if p.CurrentVersion() != 0 { t.Fatalf("env provider version: %d", p.CurrentVersion()) } b, err := p.KeyBytes(0) - if err != nil || string(b) != "env-provided-key-material" { + if err != nil || string(b) != "env-provided-key-material-at-least-32-bytes!" { t.Fatalf("KeyBytes(0) = %q, %v", b, err) } if _, err := p.KeyBytes(1); !errors.Is(err, ErrUnknownKeyVersion) { @@ -157,6 +160,54 @@ func TestEnvKeyProvider(t *testing.T) { } } +// TestEnvKeyProviderKeyPolicy covers the accepted key formats and the +// rejection of weak material: HKDF would stretch any input to 32 bytes, but a +// short password retains only the password's entropy and must be refused. +func TestEnvKeyProviderKeyPolicy(t *testing.T) { + raw32 := "0123456789abcdef0123456789abcdef" // exactly 32 bytes + p := NewEnvKeyProvider("YAAD_TEST_KEY") + + t.Setenv("YAAD_TEST_KEY", raw32) + b, err := p.KeyBytes(0) + if err != nil || string(b) != raw32 { + t.Fatalf("raw 32-byte key rejected: %q, %v", b, err) + } + + key32 := make([]byte, 32) + for i := range key32 { + key32[i] = byte(i) + } + // Encoded 32-byte keys are 43+ (base64) or 64 (hex) characters, so the + // documented "raw wins if ambiguous" tie-break applies: they are used as + // raw material rather than decoded. + for name, encoded := range map[string]string{ + "base64": base64.StdEncoding.EncodeToString(key32), + "base64url": base64.RawURLEncoding.EncodeToString(key32), + "hex": hex.EncodeToString(key32), + } { + t.Setenv("YAAD_TEST_KEY", encoded) + b, err := p.KeyBytes(0) + if err != nil || !bytes.Equal(b, []byte(encoded)) { + t.Fatalf("%s-encoded key (len %d) should be used raw: %v", name, len(encoded), err) + } + } + + for _, weak := range []string{ + "short", // far too short + "0123456789abcdef0123456789abc", // 31 bytes: one short of the raw policy + "Y2FyaXR5MTZieXRlcw==", // valid base64, but decodes to 16 bytes + "0011223344556677", // valid hex, but decodes to 8 bytes + } { + t.Setenv("YAAD_TEST_KEY", weak) + _, err := p.KeyBytes(0) + if !errors.Is(err, ErrKeyMaterial) { + t.Errorf("weak key %q accepted, want ErrKeyMaterial", weak) + } else if !strings.Contains(err.Error(), "YAAD_TEST_KEY") { + t.Errorf("error should name the env var: %v", err) + } + } +} + // TestStoreEncryptionRoundTrip covers the full storage path: writes encrypt, // reads decrypt, and the bookkeeping columns carry the right flags. func TestStoreEncryptionRoundTrip(t *testing.T) { diff --git a/storage/lock.go b/storage/lock.go index 3e04eed..b58a447 100644 --- a/storage/lock.go +++ b/storage/lock.go @@ -3,6 +3,7 @@ package storage import ( "fmt" "os" + "path/filepath" "sync" ) @@ -44,7 +45,16 @@ func AcquireProcessLock(dbPath string) (*ProcessLock, error) { procMu.Lock() defer procMu.Unlock() + // Normalize the lock path so equivalent spellings of the same database + // ("data/yaad.db", "./data/yaad.db", "x/../data/yaad.db") share one + // procHolders entry. Without this, two paths resolving to the same lock + // file would open it twice in-process; flock treats the two open file + // descriptions as separate holders, so the second open would spuriously + // fail with "locked by another yaad process". lockPath := dbPath + ".lock" + if abs, err := filepath.Abs(lockPath); err == nil { + lockPath = abs + } if p, ok := procHolders[lockPath]; ok { p.refs++ return &ProcessLock{proc: p}, nil diff --git a/storage/lock_process_test.go b/storage/lock_process_test.go index 462f228..3386993 100644 --- a/storage/lock_process_test.go +++ b/storage/lock_process_test.go @@ -125,3 +125,38 @@ func runLockChild() { fmt.Println("LOCKBLOCKED") os.Exit(0) } + +// TestProcessLockPathNormalization verifies that equivalent spellings of the +// same database path share one lock entry: without normalization, opening +// "dir/db" and "dir/./db" created two procHolders entries pointing at the +// same lock file, and the second flock failed as if another process held it. +func TestProcessLockPathNormalization(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "norm.db") + + s1, err := NewStore(dbPath) + if err != nil { + t.Fatalf("NewStore(clean path): %v", err) + } + defer s1.Close() + + // Same database through lexically unclean absolute paths (constructed + // via concatenation because filepath.Join cleans them away). + s2, err := NewStore(dir + "/./norm.db") + if err != nil { + t.Fatalf("NewStore(dot-segment path): %v", err) + } + defer s2.Close() + s3, err := NewStore(dir + "/junk/../norm.db") + if err != nil { + t.Fatalf("NewStore(parent-ref path): %v", err) + } + defer s3.Close() + + procMu.Lock() + holders := len(procHolders) + procMu.Unlock() + if holders != 1 { + t.Errorf("expected 1 lock holder after 3 equivalent paths, got %d", holders) + } +} diff --git a/storage/metadata_test.go b/storage/metadata_test.go index e474fc4..fecab96 100644 --- a/storage/metadata_test.go +++ b/storage/metadata_test.go @@ -2,6 +2,7 @@ package storage import ( "context" + "fmt" "testing" ) @@ -148,3 +149,52 @@ func TestFillNodeMetadataEmptyIsNoop(t *testing.T) { t.Errorf("FillNodeMetadata(nil): %v", err) } } + +// TestLoadNodeMetadataChunks is a regression test for the missing chunking in +// LoadNodeMetadata: it used to build a single IN (...) with every ID, which +// exceeds SQLite's host-parameter limit (maxSQLVariables = 900) once the ID +// set grew large enough. More than 900 IDs must load without error. +func TestLoadNodeMetadataChunks(t *testing.T) { + s, cleanup := setupStore(t) + defer cleanup() + ctx := context.Background() + + const total = maxSQLVariables + 50 // forces at least two chunks + ids := make([]string, total) + for i := 0; i < total; i++ { + ids[i] = fmt.Sprintf("meta-chunk-%d", i) + if err := s.CreateNode(ctx, &Node{ID: ids[i], Type: "convention", Content: ids[i], ContentHash: "h" + ids[i], Scope: "project"}); err != nil { + t.Fatalf("CreateNode(%s): %v", ids[i], err) + } + if err := s.SaveNodeMetadata(ctx, ids[i], map[string]string{"idx": fmt.Sprintf("%d", i)}); err != nil { + t.Fatalf("SaveNodeMetadata(%s): %v", ids[i], err) + } + } + + loaded, err := s.LoadNodeMetadata(ctx, ids) + if err != nil { + t.Fatalf("LoadNodeMetadata with %d ids: %v", total, err) + } + if len(loaded) != total { + t.Fatalf("expected metadata for %d nodes, got %d", total, len(loaded)) + } + // Spot-check the first and last IDs (they land in different chunks). + if got := loaded[ids[0]]["idx"]; got != "0" { + t.Errorf("first node metadata idx = %q, want 0", got) + } + if got := loaded[ids[total-1]]["idx"]; got != fmt.Sprintf("%d", total-1) { + t.Errorf("last node metadata idx = %q, want %d", got, total-1) + } + + // FillNodeMetadata drives the same query path. + nodes := make([]*Node, total) + for i, id := range ids { + nodes[i] = &Node{ID: id} + } + if err := s.FillNodeMetadata(ctx, nodes); err != nil { + t.Fatalf("FillNodeMetadata with %d nodes: %v", total, err) + } + if nodes[0].Metadata["idx"] != "0" || nodes[total-1].Metadata["idx"] != fmt.Sprintf("%d", total-1) { + t.Errorf("FillNodeMetadata chunk merge mismatch: first=%v last=%v", nodes[0].Metadata, nodes[total-1].Metadata) + } +} diff --git a/storage/sqlite.go b/storage/sqlite.go index 6dd6bdd..b6f4409 100644 --- a/storage/sqlite.go +++ b/storage/sqlite.go @@ -287,7 +287,28 @@ func NewStore(dbPath string) (*Store, error) { // parameters were silently ignored — the database was not in WAL mode // and the busy timeout was effectively 0, causing all SQLITE_BUSY // failures the audit fixes had worked around. - dsn := fmt.Sprintf("%s?_pragma=journal_mode(WAL)&_pragma=foreign_keys(ON)&_pragma=busy_timeout(%d)", dbPath, defaultBusyTimeoutMs) + // + // The driver applies every _pragma parameter on EACH new pooled + // connection (driver applyQueryParams), which is where per-connection + // settings must live: synchronous, temp_store, cache_size, mmap_size, + // wal_autocheckpoint, and recursive_triggers issued once through + // db.ExecContext would only configure whichever single connection + // happened to serve that call, leaving the rest of the pool (5 conns) + // on defaults. page_size is deliberately absent: it cannot change after + // the database file is created. PRAGMA optimize runs at Close instead + // (it is designed to run near connection close, not at startup). + pragmas := []string{ + "journal_mode(WAL)", + "foreign_keys(ON)", + fmt.Sprintf("busy_timeout(%d)", defaultBusyTimeoutMs), + "synchronous(NORMAL)", + "wal_autocheckpoint(1000)", + "temp_store(MEMORY)", + "mmap_size(268435456)", // 256MB + "cache_size(-32768)", // 32MB + "recursive_triggers(ON)", + } + dsn := dbPath + "?_pragma=" + strings.Join(pragmas, "&_pragma=") db, err := sql.Open("sqlite", dsn) if err != nil { return nil, err @@ -309,9 +330,6 @@ func NewStore(dbPath string) (*Store, error) { // handles brief write contention. WAL mode ensures readers never block. db.SetMaxOpenConns(5) db.SetMaxIdleConns(5) - if err := applyPragmas(context.Background(), db); err != nil { - return nil, err - } s := &Store{db: db, dbPath: dbPath, queryTimeout: 30 * time.Second, hnswIndexes: make(map[string]*HNSWIndex)} if err := s.createTables(); err != nil { return nil, err @@ -340,27 +358,6 @@ func NewStore(dbPath string) (*Store, error) { return s, nil } -// applyPragmas applies recommended SQLite PRAGMAs for performance and safety. -func applyPragmas(ctx context.Context, db *sql.DB) error { - pragmas := []string{ - "PRAGMA synchronous = NORMAL", - "PRAGMA wal_autocheckpoint = 1000", - "PRAGMA temp_store = MEMORY", - "PRAGMA mmap_size = 268435456", // 256MB - "PRAGMA page_size = 4096", - "PRAGMA cache_size = -32768", // 32MB - "PRAGMA recursive_triggers = ON", - "PRAGMA foreign_keys = ON", - "PRAGMA optimize", // auto-analyze - } - for _, p := range pragmas { - if _, err := db.ExecContext(ctx, p); err != nil { - return fmt.Errorf("pragma %s: %w", p, err) - } - } - return nil -} - // isMemoryDSN reports whether dbPath refers to an in-memory SQLite database // (no on-disk file/directory to create or lock down). func isMemoryDSN(dbPath string) bool { @@ -377,6 +374,11 @@ func (s *Store) Close() error { if s.cache != nil { _ = s.cache.close() // best-effort; closing the db below frees statements anyway } + // PRAGMA optimize updates SQLite's internal statistics so future queries + // pick good plans; it is designed to run near connection close, not at + // startup. It only reaches one pooled connection here, which matches the + // documented best-effort usage. Errors are ignored on the shutdown path. + _, _ = s.db.ExecContext(context.Background(), "PRAGMA optimize") if s.processLock != nil { _ = s.processLock.Release() } diff --git a/storage/sqlite_edges.go b/storage/sqlite_edges.go index abe2e7f..450fff1 100644 --- a/storage/sqlite_edges.go +++ b/storage/sqlite_edges.go @@ -291,6 +291,10 @@ func (s *Store) CountEdgesBatch(ctx context.Context, nodeIDs []string) (map[stri return countEdgesBatchQ(ctx, s.q(), nodeIDs) } +// countEdgesBatchQ counts inbound/outbound edges for nodeIDs. SQLite caps the +// number of host parameters per statement (maxSQLVariables), so large ID sets +// are counted chunk by chunk; each chunk's GROUP BY totals merge into the +// same per-node counts. func countEdgesBatchQ(ctx context.Context, q queryable, nodeIDs []string) (map[string][2]int, error) { if len(nodeIDs) == 0 { return nil, nil @@ -300,52 +304,65 @@ func countEdgesBatchQ(ctx context.Context, q queryable, nodeIDs []string) (map[s result[id] = [2]int{0, 0} } - // Count outbound edges (from_id IN (...)) - outQ := `SELECT from_id, COUNT(*) FROM edges WHERE from_id IN (` + placeholders(len(nodeIDs)) + `) GROUP BY from_id` - args := make([]any, len(nodeIDs)) - for i, id := range nodeIDs { - args[i] = id - } - rows, err := q.QueryContext(ctx, outQ, args...) - if err != nil { - return nil, fmt.Errorf("count edges batch outbound: %w", err) - } - defer func() { _ = rows.Close() }() - for rows.Next() { - var id string - var count int - if err := rows.Scan(&id, &count); err != nil { - return nil, err + for i := 0; i < len(nodeIDs); i += maxSQLVariables { + end := i + maxSQLVariables + if end > len(nodeIDs) { + end = len(nodeIDs) } - if v, ok := result[id]; ok { - v[1] = count - result[id] = v + chunk := nodeIDs[i:end] + args := make([]any, len(chunk)) + for j, id := range chunk { + args[j] = id } - } - if err := rows.Err(); err != nil { - return nil, err - } + ph := placeholders(len(chunk)) - // Count inbound edges (to_id IN (...)) - inQ := `SELECT to_id, COUNT(*) FROM edges WHERE to_id IN (` + placeholders(len(nodeIDs)) + `) GROUP BY to_id` - rows2, err := q.QueryContext(ctx, inQ, args...) - if err != nil { - return nil, fmt.Errorf("count edges batch inbound: %w", err) - } - defer func() { _ = rows2.Close() }() - for rows2.Next() { - var id string - var count int - if err := rows2.Scan(&id, &count); err != nil { + // Count outbound edges (from_id IN (...)) + outQ := `SELECT from_id, COUNT(*) FROM edges WHERE from_id IN (` + ph + `) GROUP BY from_id` + rows, err := q.QueryContext(ctx, outQ, args...) + if err != nil { + return nil, fmt.Errorf("count edges batch outbound: %w", err) + } + for rows.Next() { + var id string + var count int + if err := rows.Scan(&id, &count); err != nil { + _ = rows.Close() + return nil, err + } + if v, ok := result[id]; ok { + v[1] = count + result[id] = v + } + } + if err := rows.Err(); err != nil { + _ = rows.Close() return nil, err } - if v, ok := result[id]; ok { - v[0] = count - result[id] = v + _ = rows.Close() + + // Count inbound edges (to_id IN (...)) + inQ := `SELECT to_id, COUNT(*) FROM edges WHERE to_id IN (` + ph + `) GROUP BY to_id` + rows2, err := q.QueryContext(ctx, inQ, args...) + if err != nil { + return nil, fmt.Errorf("count edges batch inbound: %w", err) } - } - if err := rows2.Err(); err != nil { - return nil, err + for rows2.Next() { + var id string + var count int + if err := rows2.Scan(&id, &count); err != nil { + _ = rows2.Close() + return nil, err + } + if v, ok := result[id]; ok { + v[0] = count + result[id] = v + } + } + if err := rows2.Err(); err != nil { + _ = rows2.Close() + return nil, err + } + _ = rows2.Close() } return result, nil diff --git a/storage/sqlite_nodes.go b/storage/sqlite_nodes.go index 7dba1b4..7abbfb2 100644 --- a/storage/sqlite_nodes.go +++ b/storage/sqlite_nodes.go @@ -221,9 +221,34 @@ func (s *Store) DeleteNode(ctx context.Context, id string) error { }, 5, 50*time.Millisecond) } +// deleteNodeQ removes a node and every dependent row, in one transaction +// context (both Store.DeleteNode and txStore.DeleteNode route through here). +// +// The schema declares child tables WITHOUT ON DELETE CASCADE — embeddings, +// file_watch, and node_signatures reference nodes(id) directly, and +// embeddings_hnsw references embeddings(node_id) — so with foreign_keys(ON) +// a bare `DELETE FROM nodes` fails with an FK violation whenever any child +// row exists. node_versions has no FK at all and would silently leak orphans. +// Delete every child explicitly, leaf tables first, before the node itself. func deleteNodeQ(ctx context.Context, q queryable, id string) error { - if _, err := q.ExecContext(ctx, `DELETE FROM edges WHERE from_id=? OR to_id=?`, id, id); err != nil { - return err + // Order matters: embeddings_hnsw must go before embeddings, and all + // node-referencing children before nodes. + children := []struct { + query string + args []any + }{ + {`DELETE FROM edges WHERE from_id=? OR to_id=?`, []any{id, id}}, + {`DELETE FROM node_signatures WHERE node_id=?`, []any{id}}, + {`DELETE FROM file_watch WHERE node_id=?`, []any{id}}, + {`DELETE FROM embeddings_hnsw WHERE node_id=?`, []any{id}}, + {`DELETE FROM embeddings WHERE node_id=?`, []any{id}}, + {`DELETE FROM node_versions WHERE node_id=?`, []any{id}}, + {`DELETE FROM node_metadata WHERE node_id=?`, []any{id}}, // cascade-deleted anyway; explicit for FK-off safety + } + for _, c := range children { + if _, err := q.ExecContext(ctx, c.query, c.args...); err != nil { + return err + } } _, err := q.ExecContext(ctx, `DELETE FROM nodes WHERE id=?`, id) return err @@ -575,44 +600,51 @@ func (s *Store) LoadNodeMetadata(ctx context.Context, nodeIDs []string) (map[str if len(nodeIDs) == 0 { return nil, nil } - ids, args := chunkedArgs(nodeIDs) return retryOnBusyVal(func() (map[string]map[string]string, error) { - return loadNodeMetadataQ(ctx, s.q(), ids, args) + ctx, cancel := s.withTimeout(ctx) + defer cancel() + return loadNodeMetadataQ(ctx, s.q(), nodeIDs) }, 2, 10*time.Millisecond) } -func loadNodeMetadataQ(ctx context.Context, q queryable, idsChunk []string, args []any) (map[string]map[string]string, error) { - query := "SELECT node_id, key, value FROM node_metadata WHERE node_id IN (?" - for i := 1; i < len(idsChunk); i++ { - query += ", ?" - } - query += ")" - rows, err := q.QueryContext(ctx, query, args...) - if err != nil { - return nil, err - } - defer func() { _ = rows.Close() }() - +// loadNodeMetadataQ fetches node_metadata rows for nodeIDs. SQLite caps the +// number of host parameters per statement (maxSQLVariables), so large ID sets +// are queried in chunks and the per-node results merged. +func loadNodeMetadataQ(ctx context.Context, q queryable, nodeIDs []string) (map[string]map[string]string, error) { result := make(map[string]map[string]string) - for rows.Next() { - var nodeID, key, value string - if err := rows.Scan(&nodeID, &key, &value); err != nil { + for i := 0; i < len(nodeIDs); i += maxSQLVariables { + end := i + maxSQLVariables + if end > len(nodeIDs) { + end = len(nodeIDs) + } + chunk := nodeIDs[i:end] + query := "SELECT node_id, key, value FROM node_metadata WHERE node_id IN (" + placeholders(len(chunk)) + ")" + args := make([]any, len(chunk)) + for j, id := range chunk { + args[j] = id + } + rows, err := q.QueryContext(ctx, query, args...) + if err != nil { return nil, err } - if result[nodeID] == nil { - result[nodeID] = make(map[string]string) + for rows.Next() { + var nodeID, key, value string + if err := rows.Scan(&nodeID, &key, &value); err != nil { + _ = rows.Close() + return nil, err + } + if result[nodeID] == nil { + result[nodeID] = make(map[string]string) + } + result[nodeID][key] = value + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err } - result[nodeID][key] = value - } - return result, rows.Err() -} - -func chunkedArgs(ids []string) ([]string, []any) { - args := make([]any, len(ids)) - for i, id := range ids { - args[i] = id + _ = rows.Close() } - return ids, args + return result, nil } // FillNodeMetadata loads metadata for all given nodes in one batch query and diff --git a/storage/sqlite_test.go b/storage/sqlite_test.go index 7967ac6..ac8b43a 100644 --- a/storage/sqlite_test.go +++ b/storage/sqlite_test.go @@ -2,6 +2,7 @@ package storage import ( "context" + "database/sql" "errors" "fmt" "os" @@ -614,3 +615,184 @@ func TestHNSWIndex(t *testing.T) { t.Fatalf("expected 2 results for model2, got %d", len(ids)) } } + +// TestDeleteNodeRemovesChildren is a regression test for the FK-violation bug +// where deleteNodeQ only deleted edges + nodes. With foreign_keys(ON) and no +// ON DELETE CASCADE on embeddings / file_watch / node_signatures, deleting a +// node with child rows used to fail; node_versions and node_metadata rows +// leaked orphans. DeleteNode must remove every child and succeed. +func TestDeleteNodeRemovesChildren(t *testing.T) { + s, cleanup := setupStore(t) + defer cleanup() + ctx := context.Background() + + a := &Node{ + ID: "node-child-a", + Type: "convention", + Content: "content A", + ContentHash: "hash-a", + Scope: "project", + Project: "test", + Metadata: map[string]string{"k": "v"}, + } + b := &Node{ID: "node-child-b", Type: "convention", Content: "content B", ContentHash: "hash-b", Scope: "project", Project: "test"} + if err := s.CreateNode(ctx, a); err != nil { + t.Fatalf("CreateNode a: %v", err) + } + if err := s.CreateNode(ctx, b); err != nil { + t.Fatalf("CreateNode b: %v", err) + } + + // Populate every child table that references the node. + if err := s.CreateEdge(ctx, &Edge{ID: "e1", FromID: a.ID, ToID: b.ID, Type: "relates"}); err != nil { + t.Fatalf("CreateEdge: %v", err) + } + if err := s.SaveEmbedding(ctx, a.ID, "default", []float32{1, 0, 0}); err != nil { + t.Fatalf("SaveEmbedding: %v", err) + } + if err := s.SaveSignature(ctx, a.ID, "sig-a"); err != nil { + t.Fatalf("SaveSignature: %v", err) + } + if err := s.AddFileWatch(ctx, "/tmp/file.go", a.ID, "githash"); err != nil { + t.Fatalf("AddFileWatch: %v", err) + } + if err := s.SaveVersion(ctx, a.ID, "v1 content", "tester", "regression"); err != nil { + t.Fatalf("SaveVersion: %v", err) + } + // embeddings_hnsw rows are normally written by the HNSW index persist + // path; insert one directly to prove the FK chain is honored. + if _, err := s.db.ExecContext(ctx, + `INSERT INTO embeddings_hnsw (node_id, vector, model, neighbors, updated_at) VALUES (?, ?, 'default', '[]', CURRENT_TIMESTAMP)`, + a.ID, EncodeVector([]float32{1, 0, 0})); err != nil { + t.Fatalf("insert embeddings_hnsw: %v", err) + } + + if err := s.DeleteNode(ctx, a.ID); err != nil { + t.Fatalf("DeleteNode with children: %v", err) + } + + if _, err := s.GetNode(ctx, a.ID); err == nil { + t.Error("node still present after DeleteNode") + } + for _, tbl := range []string{"edges", "embeddings", "embeddings_hnsw", "file_watch", "node_signatures", "node_versions", "node_metadata"} { + var n int + col := "node_id" + if tbl == "edges" { + col = "from_id" + } + if err := s.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+tbl+" WHERE "+col+"=?", a.ID).Scan(&n); err != nil { + t.Fatalf("count %s: %v", tbl, err) + } + if n != 0 { + t.Errorf("table %s still has %d rows for deleted node", tbl, n) + } + } + + // The other endpoint of the edge must survive untouched. + if _, err := s.GetNode(ctx, b.ID); err != nil { + t.Errorf("unrelated node b deleted: %v", err) + } +} + +// TestPerConnectionPragmas verifies the per-connection pragmas (synchronous, +// cache_size, temp_store, foreign_keys, recursive_triggers) are configured on +// EVERY pooled connection, not just the first one opened. Each open +// transaction pins a distinct connection, so holding MaxOpenOpenConns(5) +// transactions at once forces the checks below onto all five connections. +func TestPerConnectionPragmas(t *testing.T) { + s, cleanup := setupStore(t) + defer cleanup() + ctx := context.Background() + + const pool = 5 + txs := make([]*sql.Tx, 0, pool) + defer func() { + for _, tx := range txs { + _ = tx.Rollback() + } + }() + for i := 0; i < pool; i++ { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("begin tx %d: %v", i, err) + } + txs = append(txs, tx) + } + + checks := []struct { + pragma string + want int + }{ + {"synchronous", 1}, // NORMAL + {"cache_size", -32768}, // 32MB + {"temp_store", 2}, // MEMORY + {"foreign_keys", 1}, // ON + {"recursive_triggers", 1}, // ON + {"wal_autocheckpoint", 1000}, + } + for i, tx := range txs { + for _, c := range checks { + var got int + if err := tx.QueryRowContext(ctx, "PRAGMA "+c.pragma).Scan(&got); err != nil { + t.Fatalf("conn %d: PRAGMA %s: %v", i, c.pragma, err) + } + if got != c.want { + t.Errorf("conn %d: PRAGMA %s = %d, want %d", i, c.pragma, got, c.want) + } + } + } +} + +// TestCountEdgesBatchChunks is a regression test for the missing chunking in +// CountEdgesBatch: it built a single IN (...) with every node ID, exceeding +// SQLite's host-parameter limit (maxSQLVariables = 900) for large ID sets. +// Edges deliberately span the chunk boundary and cross directions. +func TestCountEdgesBatchChunks(t *testing.T) { + s, cleanup := setupStore(t) + defer cleanup() + ctx := context.Background() + + const total = maxSQLVariables + 50 // forces at least two chunks + ids := make([]string, total) + for i := 0; i < total; i++ { + ids[i] = fmt.Sprintf("count-chunk-%d", i) + if err := s.CreateNode(ctx, &Node{ID: ids[i], Type: "convention", Content: ids[i], ContentHash: "h" + ids[i], Scope: "project"}); err != nil { + t.Fatalf("CreateNode(%s): %v", ids[i], err) + } + } + // Outbound edges from the first node (chunk 1) to nodes in chunk 2, and + // inbound edges into the last node from nodes in chunk 1. + for i := maxSQLVariables - 2; i < maxSQLVariables+2; i++ { + e := &Edge{ID: fmt.Sprintf("e-out-%d", i), FromID: ids[0], ToID: ids[i], Type: "relates"} + if err := s.CreateEdge(ctx, e); err != nil { + t.Fatalf("CreateEdge out %d: %v", i, err) + } + e2 := &Edge{ID: fmt.Sprintf("e-in-%d", i), FromID: ids[i], ToID: ids[total-1], Type: "relates"} + if err := s.CreateEdge(ctx, e2); err != nil { + t.Fatalf("CreateEdge in %d: %v", i, err) + } + } + + counts, err := s.CountEdgesBatch(ctx, ids) + if err != nil { + t.Fatalf("CountEdgesBatch with %d ids: %v", total, err) + } + if len(counts) != total { + t.Fatalf("expected counts for %d nodes, got %d", total, len(counts)) + } + if got := counts[ids[0]][1]; got != 4 { + t.Errorf("first node outbound = %d, want 4", got) + } + if got := counts[ids[0]][0]; got != 0 { + t.Errorf("first node inbound = %d, want 0", got) + } + if got := counts[ids[total-1]][0]; got != 4 { + t.Errorf("last node inbound = %d, want 4", got) + } + if got := counts[ids[total-1]][1]; got != 0 { + t.Errorf("last node outbound = %d, want 0", got) + } + if got := counts[ids[10]]; got != [2]int{0, 0} { + t.Errorf("isolated node counts = %v, want zeros", got) + } +} diff --git a/storage/sqlite_tx.go b/storage/sqlite_tx.go index 22a5c63..09ee5bc 100644 --- a/storage/sqlite_tx.go +++ b/storage/sqlite_tx.go @@ -255,8 +255,7 @@ func (t *txStore) SaveNodeMetadata(ctx context.Context, nodeID string, meta map[ } func (t *txStore) LoadNodeMetadata(ctx context.Context, nodeIDs []string) (map[string]map[string]string, error) { - ids, args := chunkedArgs(nodeIDs) - return loadNodeMetadataQ(ctx, t.tx, ids, args) + return loadNodeMetadataQ(ctx, t.tx, nodeIDs) } func (t *txStore) SaveSignature(ctx context.Context, nodeID, signature string) error {