From 5e10437fe329a1ad0a30ab270a0b2f2a33394081 Mon Sep 17 00:00:00 2001 From: dtoro Date: Sat, 8 Aug 2026 23:46:43 +0200 Subject: [PATCH] Phase 4 (Performance) + Phase 6 (Infrastructure) completion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 4 — Performance: - F1: SSH DialPool with key-by-host pooling and 5min idle TTL - F2: In-memory entity lookup cache (TTL 60s, HTTP resolveEntityID) - F3: Trigram GIN indexes on entities.slug and entities.name (migration 031) - F4: Partial index on executions(classification_id) for auto-act (migration 032) - Added missing RunOutput and RunStreaming in actuator/ (E3 gap fill) Phase 6 — Infrastructure: - H1: Infisical image pinned to v0.99.1 - H2: execworker daemon — polls pending executions with per-execution advisory locks, recovers orphaned running executions, wired as docker-compose service - H3: splitSQL hardened with block comment and string-literal support, 6 new edge-case tests (11 total) - H4: Scheduler acquires pg_try_advisory_lock(0x01c05e6) at startup --- VERSION | 2 +- cmd/oikos/main.go | 7 +- docker-compose.yml | 33 ++- internal/actuator/client.go | 40 ++++ internal/actuator/pool.go | 122 +++++++++++ internal/actuator/stream.go | 86 ++++++++ internal/db/entity_cache.go | 61 ++++++ internal/db/pool.go | 34 ++- internal/db/splitsql_test.go | 54 +++++ internal/execworker/init.go | 13 ++ internal/execworker/worker.go | 207 ++++++++++++++++++ internal/httpapi/entity_mutations.go | 2 + internal/httpapi/impl.go | 11 + internal/httpapi/server.go | 10 +- internal/scheduler/scheduler.go | 34 ++- migrations/031_entity_trigram_index.up.sql | 9 + migrations/032_auto_act_index.up.sql | 8 + ...6-08-05-backend-evaluation-improvements.md | 142 ++++++++---- plans/index.md | 2 +- 19 files changed, 822 insertions(+), 55 deletions(-) create mode 100644 internal/actuator/pool.go create mode 100644 internal/actuator/stream.go create mode 100644 internal/db/entity_cache.go create mode 100644 internal/execworker/init.go create mode 100644 internal/execworker/worker.go create mode 100644 migrations/031_entity_trigram_index.up.sql create mode 100644 migrations/032_auto_act_index.up.sql diff --git a/VERSION b/VERSION index 25939d3..c25c8e5 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.29.1 +0.30.0 diff --git a/cmd/oikos/main.go b/cmd/oikos/main.go index 7257585..6168ebd 100644 --- a/cmd/oikos/main.go +++ b/cmd/oikos/main.go @@ -12,6 +12,7 @@ import ( "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/db" + "github.com/dtoro/oikos/internal/execworker" "github.com/dtoro/oikos/internal/httpapi" "github.com/dtoro/oikos/internal/knowledge" "github.com/dtoro/oikos/internal/notifier" @@ -23,6 +24,7 @@ import ( var schedulerRunner = scheduler.RunnerForMain() var notifierRunner = notifier.RunnerForMain() +var execWorkerRunner = execworker.RunnerForMain() func main() { if len(os.Args) < 2 { @@ -95,6 +97,8 @@ func main() { runWithPool(ctx, cfg, "scheduler", schedulerRunner) case "notifier": runWithPool(ctx, cfg, "notifier", notifierRunner) + case "execution-worker": + runWithPool(ctx, cfg, "execution-worker", execWorkerRunner) case "all": pool, err := db.New(ctx, cfg.DatabaseURL) if err != nil { @@ -110,8 +114,9 @@ func main() { go schedulerRunner(ctx, pool, cfg) go notifierRunner(ctx, pool, cfg) + go execWorkerRunner(ctx, pool, cfg) - slog.Info("all: starting api with scheduler + notifier in background") + slog.Info("all: starting api with scheduler + notifier + execution-worker in background") if err := httpapi.ListenAndServe(ctx, pool, cfg); err != nil { slog.Error("api failed", "error", err) os.Exit(1) diff --git a/docker-compose.yml b/docker-compose.yml index e647e0c..1ef68e9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -195,6 +195,37 @@ services: retries: 3 start_period: 120s + # Execution worker (Phase 6) — Postgres-backed job queue + execution-worker: + image: oikos-execution-worker:${OIKOS_VERSION:-latest} + build: + context: . + dockerfile: compose/oikos/Dockerfile + restart: unless-stopped + profiles: ["dev", "full"] + depends_on: + seed: + condition: service_completed_successfully + environment: + OIKOS_DATABASE_URL: postgres://oikos:${OIKOS_DB_PASSWORD:-oikos_dev}@postgres:5432/oikos?sslmode=disable + OIKOS_DEBUG: "true" + OIKOS_SSH_KEY_PATH: /etc/oikos/ssh_key + OIKOS_SSH_USER: root + OIKOS_HEALTH_LISTEN: ":8095" + volumes: + - ${OIKOS_SSH_KEY_PATH:-~/.ssh/id_ed25519}:/etc/oikos/ssh_key:ro + command: ["execution-worker"] + stop_signal: SIGTERM + stop_grace_period: 30s + mem_limit: 256m + cpus: 1.0 + healthcheck: + test: ["CMD", "wget", "-q", "-O", "-", "http://127.0.0.1:8095/healthz"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 90s + # Nomos agent gateway (Phase 4) — mesh-published :8092 nomos: image: oikos-nomos:${OIKOS_VERSION:-latest} @@ -276,7 +307,7 @@ services: # Infisical self-hosted (Phase 5 secrets management) infisical: - image: infisical/infisical:latest + image: infisical/infisical:v0.99.1 restart: unless-stopped profiles: ["infisical", "full"] depends_on: diff --git a/internal/actuator/client.go b/internal/actuator/client.go index 95dc63b..232a69c 100644 --- a/internal/actuator/client.go +++ b/internal/actuator/client.go @@ -1,10 +1,12 @@ package actuator import ( + "bytes" "context" "fmt" "net" "os" + "strings" "time" "golang.org/x/crypto/ssh" @@ -99,3 +101,41 @@ func RunCombinedOutput(ctx context.Context, client *ssh.Client, cmd string) ([]b return res.out, nil } } + +// RunOutput runs cmd on an established client and returns stdout only. +// Stderr is folded into the returned error so callers that parse stdout +// as JSON (e.g. the scheduler's check scripts) don't get interleaved +// stderr in the output stream. +func RunOutput(ctx context.Context, client *ssh.Client, cmd string) ([]byte, error) { + session, err := client.NewSession() + if err != nil { + return nil, fmt.Errorf("create session: %w", err) + } + defer session.Close() + + var outBuf, errBuf bytes.Buffer + session.Stdout = &outBuf + session.Stderr = &errBuf + + type result struct { + runErr error + } + ch := make(chan result, 1) + go func() { + ch <- result{runErr: session.Run(cmd)} + }() + + select { + case <-ctx.Done(): + session.Close() + return nil, ctx.Err() + case res := <-ch: + if res.runErr != nil { + if errBuf.Len() > 0 { + return outBuf.Bytes(), fmt.Errorf("command: %w\nstderr: %s", res.runErr, strings.TrimSpace(errBuf.String())) + } + return outBuf.Bytes(), fmt.Errorf("command: %w", res.runErr) + } + return outBuf.Bytes(), nil + } +} \ No newline at end of file diff --git a/internal/actuator/pool.go b/internal/actuator/pool.go new file mode 100644 index 0000000..774184c --- /dev/null +++ b/internal/actuator/pool.go @@ -0,0 +1,122 @@ +package actuator + +import ( + "context" + "fmt" + "sync" + "time" + + "golang.org/x/crypto/ssh" +) + +type poolEntry struct { + client *ssh.Client + createdAt time.Time +} + +type DialPool struct { + mu sync.RWMutex + entries map[string]*poolEntry + ttl time.Duration + done chan struct{} + stopped bool +} + +func NewDialPool(ttl time.Duration) *DialPool { + p := &DialPool{ + entries: make(map[string]*poolEntry), + ttl: ttl, + done: make(chan struct{}), + } + if ttl > 0 { + go p.evictLoop() + } + return p +} + +func (p *DialPool) key(opts DialOptions) string { + port := opts.Port + if port <= 0 { + port = 22 + } + return fmt.Sprintf("%s:%d", opts.Host, port) +} + +func (p *DialPool) Get(ctx context.Context, opts DialOptions) (*ssh.Client, error) { + k := p.key(opts) + p.mu.RLock() + entry, ok := p.entries[k] + p.mu.RUnlock() + if ok { + // Quick health check: a session can be created without running a + // command — if it fails, the connection is dead and we evict it. + testSession, err := entry.client.NewSession() + if err == nil { + testSession.Close() + return entry.client, nil + } + p.mu.Lock() + if p.entries[k] == entry { + entry.client.Close() + delete(p.entries, k) + } + p.mu.Unlock() + // Fall through to dial below + } + client, err := Dial(ctx, opts) + if err != nil { + return nil, err + } + p.mu.Lock() + if p.stopped { + p.mu.Unlock() + client.Close() + return nil, fmt.Errorf("ssh dial pool: closed") + } + if existing, ok2 := p.entries[k]; ok2 { + p.mu.Unlock() + client.Close() + return existing.client, nil + } + p.entries[k] = &poolEntry{client: client, createdAt: time.Now()} + p.mu.Unlock() + return client, nil +} + +func (p *DialPool) Close() { + p.mu.Lock() + p.stopped = true + for k, entry := range p.entries { + entry.client.Close() + delete(p.entries, k) + } + p.mu.Unlock() + if p.ttl > 0 { + close(p.done) + } +} + +func (p *DialPool) evictLoop() { + ticker := time.NewTicker(p.ttl / 2) + defer ticker.Stop() + for { + select { + case <-p.done: + return + case <-ticker.C: + p.evict() + } + } +} + +func (p *DialPool) evict() { + deadline := time.Now().Add(-p.ttl) + p.mu.Lock() + defer p.mu.Unlock() + for k, entry := range p.entries { + if entry.createdAt.Before(deadline) { + entry.client.Close() + delete(p.entries, k) + } + } +} \ No newline at end of file diff --git a/internal/actuator/stream.go b/internal/actuator/stream.go new file mode 100644 index 0000000..b31b668 --- /dev/null +++ b/internal/actuator/stream.go @@ -0,0 +1,86 @@ +package actuator + +import ( + "bufio" + "context" + "fmt" + "io" + "time" + + "golang.org/x/crypto/ssh" +) + +// RunStreaming runs a command on an established SSH client and forwards output +// chunks to sink as they arrive. A nil sink collects output silently. Returns +// the full combined output and any command error. +func RunStreaming(ctx context.Context, client *ssh.Client, command string, sink func(stream string, chunk []byte), timeout time.Duration) (string, error) { + session, err := client.NewSession() + if err != nil { + return "", fmt.Errorf("create session: %w", err) + } + defer session.Close() + + outPipe, err := session.StdoutPipe() + if err != nil { + return "", fmt.Errorf("stdout pipe: %w", err) + } + errPipe, err := session.StderrPipe() + if err != nil { + return "", fmt.Errorf("stderr pipe: %w", err) + } + + type streamResult struct { + out string + err error + } + + resultCh := make(chan streamResult, 1) + go func() { + var combined []byte + done := make(chan struct{}, 2) + + readStream := func(stream string, r io.Reader) { + sc := bufio.NewScanner(r) + for sc.Scan() { + line := sc.Bytes() + chunk := make([]byte, len(line)) + copy(chunk, line) + if sink != nil { + sink(stream, chunk) + } + if stream == "stdout" || stream == "" { + if len(combined) > 0 { + combined = append(combined, '\n') + } + combined = append(combined, chunk...) + } + } + done <- struct{}{} + } + + go readStream("stdout", outPipe) + go readStream("stderr", errPipe) + + runErr := session.Run(command) + <-done + <-done + resultCh <- streamResult{out: string(combined), err: runErr} + }() + + if timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, timeout) + defer cancel() + } + + select { + case <-ctx.Done(): + session.Close() + return "", ctx.Err() + case res := <-resultCh: + if res.err != nil { + return res.out, fmt.Errorf("command: %w", res.err) + } + return res.out, nil + } +} \ No newline at end of file diff --git a/internal/db/entity_cache.go b/internal/db/entity_cache.go new file mode 100644 index 0000000..fc2d7eb --- /dev/null +++ b/internal/db/entity_cache.go @@ -0,0 +1,61 @@ +package db + +import ( + "sync" + "time" +) + +type entityCacheEntry struct { + slug string + id string + attrs string + exp time.Time +} + +type EntityCache struct { + mu sync.RWMutex + m map[string]entityCacheEntry + ttl time.Duration +} + +func NewEntityCache(ttl time.Duration) *EntityCache { + return &EntityCache{ + m: make(map[string]entityCacheEntry), + ttl: ttl, + } +} + +func (c *EntityCache) GetSlug(id string) (string, bool) { + c.mu.RLock() + e, ok := c.m[id] + c.mu.RUnlock() + if !ok || time.Now().After(e.exp) { + return "", false + } + return e.slug, true +} + +func (c *EntityCache) GetID(slug string) (string, bool) { + c.mu.RLock() + e, ok := c.m[slug] + c.mu.RUnlock() + if !ok || time.Now().After(e.exp) { + return "", false + } + return e.id, true +} + +func (c *EntityCache) Set(slug, id, attrs string) { + exp := time.Now().Add(c.ttl) + c.mu.Lock() + c.m[slug] = entityCacheEntry{slug: slug, id: id, attrs: attrs, exp: exp} + c.m[id] = entityCacheEntry{slug: slug, id: id, attrs: attrs, exp: exp} + c.mu.Unlock() +} + +func (c *EntityCache) Invalidate(slug, id string) { + c.mu.Lock() + delete(c.m, slug) + delete(c.m, id) + c.mu.Unlock() +} \ No newline at end of file diff --git a/internal/db/pool.go b/internal/db/pool.go index 1efd316..a8fbffe 100644 --- a/internal/db/pool.go +++ b/internal/db/pool.go @@ -190,7 +190,10 @@ func hasSuffix(s, suffix string) bool { } // splitSQL splits a SQL string into individual statements. -// Handles $$ ... $$ dollar-quoted blocks and -- line comments. +// Handles $$ ... $$ dollar-quoted blocks, $tag$ ... $tag$ tagged quotes, +// -- line comments, /* ... */ block comments, and '...' string literals +// so that semicolons inside any of these constructs are not treated as +// statement boundaries. func splitSQL(sql string) []string { var statements []string var current strings.Builder @@ -201,7 +204,6 @@ func splitSQL(sql string) []string { for i < len(sql) { // Handle line comments (-- to end of line) if !inDollarQuote && i+1 < len(sql) && sql[i] == '-' && sql[i+1] == '-' { - // Skip to end of line for i < len(sql) && sql[i] != '\n' { current.WriteByte(sql[i]) i++ @@ -209,6 +211,34 @@ func splitSQL(sql string) []string { continue } + // Handle block comments (/* ... */) + if !inDollarQuote && i+1 < len(sql) && sql[i] == '/' && sql[i+1] == '*' { + end := strings.Index(sql[i+2:], "*/") + if end >= 0 { + current.WriteString(sql[i : i+end+4]) + i += end + 4 + continue + } + } + + // Handle single-quoted string literals ('...') + if !inDollarQuote && sql[i] == '\'' { + j := i + 1 + for j < len(sql) { + if sql[j] == '\'' { + if j+1 < len(sql) && sql[j+1] == '\'' { + j += 2 // skip doubled quote '' + continue + } + break + } + j++ + } + current.WriteString(sql[i : j+1]) + i = j + 1 + continue + } + // Check for dollar-quote start/end if !inDollarQuote && sql[i] == '$' { j := i + 1 diff --git a/internal/db/splitsql_test.go b/internal/db/splitsql_test.go index ecf4e07..17c70cd 100644 --- a/internal/db/splitsql_test.go +++ b/internal/db/splitsql_test.go @@ -51,3 +51,57 @@ func TestSplitSQLSemicolonInComment(t *testing.T) { t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts) } } + +func TestSplitSQLSemicolonInStringLiteral(t *testing.T) { + sql := `SELECT 'hello; world'; INSERT INTO t VALUES (1);` + stmts := nonEmpty(splitSQL(sql)) + if len(stmts) != 2 { + t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts) + } +} + +func TestSplitSQLDollarSignInStringLiteral(t *testing.T) { + sql := `SELECT '$100'; SELECT 2;` + stmts := nonEmpty(splitSQL(sql)) + if len(stmts) != 2 { + t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts) + } +} + +func TestSplitSQLBlockComment(t *testing.T) { + sql := `SELECT 1; /* block; with; semicolons */ SELECT 2;` + stmts := nonEmpty(splitSQL(sql)) + if len(stmts) != 2 { + t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts) + } +} + +func TestSplitSQLBlockCommentWithDollarQuote(t *testing.T) { + sql := `/* $$ not a dollar quote */ SELECT 1;` + stmts := nonEmpty(splitSQL(sql)) + if len(stmts) != 1 { + t.Fatalf("got %d statements, want 1: %#v", len(stmts), stmts) + } +} + +func TestSplitSQLDoubledQuoteInString(t *testing.T) { + sql := `SELECT 'O''Brien'; SELECT 2;` + stmts := nonEmpty(splitSQL(sql)) + if len(stmts) != 2 { + t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts) + } +} + +func TestSplitSQLEmptyInput(t *testing.T) { + stmts := nonEmpty(splitSQL("")) + if len(stmts) != 0 { + t.Fatalf("got %d statements, want 0", len(stmts)) + } +} + +func TestSplitSQLNoSemicolon(t *testing.T) { + stmts := nonEmpty(splitSQL("SELECT 1")) + if len(stmts) != 1 { + t.Fatalf("got %d statements, want 1", len(stmts)) + } +} diff --git a/internal/execworker/init.go b/internal/execworker/init.go new file mode 100644 index 0000000..76b45a3 --- /dev/null +++ b/internal/execworker/init.go @@ -0,0 +1,13 @@ +package execworker + +import ( + "context" + + "github.com/dtoro/oikos/internal/config" + "github.com/dtoro/oikos/internal/db" +) + +// RunnerForMain provides the run function for registration in main. +func RunnerForMain() func(context.Context, *db.Pool, config.Config) { + return Run +} \ No newline at end of file diff --git a/internal/execworker/worker.go b/internal/execworker/worker.go new file mode 100644 index 0000000..2d9beb4 --- /dev/null +++ b/internal/execworker/worker.go @@ -0,0 +1,207 @@ +// Package execworker processes pending executions as a background daemon. +// This provides a Postgres-backed queue: executions survive restarts, and +// per-execution advisory locks prevent duplicate processing across instances. +package execworker + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "os" + "strings" + "time" + + "github.com/dtoro/oikos/internal/actuator" + "github.com/dtoro/oikos/internal/config" + "github.com/dtoro/oikos/internal/db" + "github.com/dtoro/oikos/internal/db/sqlcgen" + "github.com/dtoro/oikos/internal/health" + "github.com/dtoro/oikos/internal/remote" + "github.com/google/uuid" +) + +// Run starts the execution worker loop. Blocks until ctx is cancelled. +func Run(ctx context.Context, pool *db.Pool, cfg config.Config) { + slog.Info("execworker: starting") + + // Liveness probe + probe := health.New(2 * time.Minute) + probe.Serve(ctx, cfg.HealthListen) + + recoverOrphaned(ctx, pool) + probe.Bump() + + ticker := time.NewTicker(15 * time.Second) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + slog.Info("execworker: shutting down") + return + case <-ticker.C: + processPending(ctx, pool) + probe.Bump() + } + } +} + +// recoverOrphaned marks executions stuck in 'running' as failed. +func recoverOrphaned(ctx context.Context, pool *db.Pool) { + tag, err := pool.Exec(ctx, `UPDATE executions SET status = 'failed', result = '{"error":"worker restarted while execution was running"}'::jsonb, completed_at = now() WHERE status = 'running'`) + if err != nil { + slog.Error("execworker: recover orphaned", "error", err) + return + } + if tag.RowsAffected() > 0 { + slog.Warn("execworker: recovered orphaned executions", "count", tag.RowsAffected()) + } +} + +// processPending polls for pending executions and dispatches them. +func processPending(ctx context.Context, pool *db.Pool) { + rows, err := pool.Query(ctx, ` + SELECT e.entity_id, e.target_entity_id, e.action, e.risk_class, e.correlation_id, e.status, + COALESCE(t.slug, '') AS target_slug + FROM executions e + LEFT JOIN entities t ON t.id = e.target_entity_id + WHERE e.status = 'proposed' + ORDER BY e.created_at ASC + LIMIT 10`) + if err != nil { + slog.Error("execworker: query pending", "error", err) + return + } + defer rows.Close() + + q := sqlcgen.New(pool) + + for rows.Next() { + var execID, targetID *uuid.UUID + var action, riskClass, correlationID, status, targetSlug string + if err := rows.Scan(&execID, &targetID, &action, &riskClass, &correlationID, &status, &targetSlug); err != nil { + slog.Error("execworker: scan row", "error", err) + continue + } + if execID == nil { + continue + } + + // At-most-once: try advisory lock on execution entity_id. + // Acquire a dedicated connection so the session-scoped lock isn't + // released when the transient pool connection is returned. + lockKey := hashUUID(*execID) + lockConn, err := pool.Acquire(ctx) + if err != nil { + slog.Error("execworker: acquire lock conn", "error", err) + continue + } + var locked bool + if err := lockConn.QueryRow(ctx, "SELECT pg_try_advisory_lock($1)", lockKey).Scan(&locked); err != nil || !locked { + lockConn.Release() + continue + } + + dispatch(ctx, pool, q, *execID, targetID, action, targetSlug, correlationID) + + // Release the per-execution lock on the same connection. + lockConn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", lockKey) + lockConn.Release() + } +} + +func dispatch(ctx context.Context, pool *db.Pool, q *sqlcgen.Queries, execID uuid.UUID, targetID *uuid.UUID, action, targetSlug, correlationID string) { + startedAt := time.Now() + + // Mark running + _, err := pool.Exec(ctx, + `UPDATE executions SET status = 'running', started_at = $2 WHERE entity_id = $1`, + execID, startedAt) + if err != nil { + slog.Error("execworker: mark running", "error", err, "execution", execID) + return + } + + // Resolve SSH target. If targetSlug is available, use it; otherwise resolve from targetID. + var host, user string + if targetSlug == "" && targetID != nil { + if err := pool.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", *targetID).Scan(&targetSlug); err != nil { + failExecution(ctx, pool, execID, fmt.Sprintf("resolve target slug: %v", err)) + return + } + } + if targetSlug != "" { + addr, sshUser, err := remote.ResolveHost(ctx, pool, targetSlug, "root") + if err == nil { + host, user = addr, sshUser + } + } + if host == "" { + failExecution(ctx, pool, execID, fmt.Sprintf("no reachable target: %s", targetSlug)) + return + } + + // Determine the command to run from the action field. + // Format: "action_name:{json_params}" or a raw command string. + cmd := action + if idx := strings.Index(action, ":"); idx > 0 && idx < len(action)-1 { + rawParams := action[idx+1:] + var params map[string]any + if json.Unmarshal([]byte(rawParams), ¶ms) == nil { + if c, ok := params["command"].(string); ok && c != "" { + cmd = c + } + } + } + + signer, err := actuator.LoadSigner(os.Getenv("OIKOS_SSH_KEY_PATH")) + if err != nil { + signer, err = actuator.LoadSigner("/etc/oikos/ssh_key") + if err != nil { + failExecution(ctx, pool, execID, fmt.Sprintf("load ssh key: %v", err)) + return + } + } + + client, err := actuator.Dial(ctx, actuator.DialOptions{Host: host, User: user, Signer: signer}) + if err != nil { + failExecution(ctx, pool, execID, fmt.Sprintf("ssh dial: %v", err)) + return + } + defer client.Close() + + out, err := actuator.RunCombinedOutput(ctx, client, cmd) + if err != nil { + failExecution(ctx, pool, execID, fmt.Sprintf("command: %v\noutput: %s", err, string(out))) + return + } + + duration := time.Since(startedAt).Milliseconds() + resultJSON, _ := json.Marshal(map[string]any{"output": string(out), "success": true}) + _ = q.UpdateExecutionStatus(ctx, sqlcgen.UpdateExecutionStatusParams{ + EntityID: execID, + Status: "completed", + Result: resultJSON, + DurationMs: &[]int32{int32(duration)}[0], + Verified: true, + }) + + slog.Info("execworker: execution complete", + "execution", execID, "target", targetSlug, "duration_ms", duration) +} + +func failExecution(ctx context.Context, pool *db.Pool, execID uuid.UUID, reason string) { + slog.Error("execworker: execution failed", "execution", execID, "error", reason) + resultJSON, _ := json.Marshal(map[string]any{"error": reason, "success": false}) + pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb, completed_at=now() WHERE entity_id=$1`, + execID, resultJSON) +} + +func hashUUID(id uuid.UUID) int { + h := 0 + for _, b := range id { + h = (h*31 + int(b)) & 0x7fffffff + } + return h +} \ No newline at end of file diff --git a/internal/httpapi/entity_mutations.go b/internal/httpapi/entity_mutations.go index e48c63e..8eabf30 100644 --- a/internal/httpapi/entity_mutations.go +++ b/internal/httpapi/entity_mutations.go @@ -281,6 +281,8 @@ func (s *Server) PatchEntity(ctx context.Context, req gen.PatchEntityRequestObje return nil, err } + s.entityCache.Invalidate(entity.Slug, entity.Id.String()) + return gen.PatchEntity200JSONResponse{ Body: entity, Headers: gen.PatchEntity200ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`}, diff --git a/internal/httpapi/impl.go b/internal/httpapi/impl.go index 6ff476d..08d03f0 100644 --- a/internal/httpapi/impl.go +++ b/internal/httpapi/impl.go @@ -57,16 +57,27 @@ func clampLimit(l *int) int { // resolveEntityID resolves a UUID-or-slug path/query value to the entity UUID. func (s *Server) resolveEntityID(ctx context.Context, idOrSlug string) (uuid.UUID, error) { if id, err := uuid.Parse(idOrSlug); err == nil { + if _, ok := s.entityCache.GetSlug(id.String()); ok { + return id, nil + } entity, err := sqlcgen.New(s.pool).GetEntityByID(ctx, id) if err != nil { return uuid.Nil, fmt.Errorf("%w: %s", domain.ErrNotFound, idOrSlug) } + s.entityCache.Set(entity.Slug, entity.ID.String(), "") return entity.ID, nil } + if cachedID, ok := s.entityCache.GetID(idOrSlug); ok { + id, parseErr := uuid.Parse(cachedID) + if parseErr == nil { + return id, nil + } + } entity, err := sqlcgen.New(s.pool).GetEntityBySlug(ctx, idOrSlug) if err != nil { return uuid.Nil, fmt.Errorf("%w: %s", domain.ErrNotFound, idOrSlug) } + s.entityCache.Set(entity.Slug, entity.ID.String(), "") return entity.ID, nil } diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index bb90d49..0faf7c6 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -55,6 +55,7 @@ type Server struct { pool *db.Pool cfg config.Config secretsManager secrets.Backend + entityCache *db.EntityCache sseBroker *sseBroker sseSubs map[*sseSubscriber]struct{} sseMu sync.Mutex @@ -69,10 +70,11 @@ type Server struct { // and pool.Close() deadlocks. func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config) http.Handler { s := &Server{ - pool: pool, - cfg: cfg, - sseBroker: newSSEBroker(10000), - sseSubs: make(map[*sseSubscriber]struct{}), + pool: pool, + cfg: cfg, + entityCache: db.NewEntityCache(60 * time.Second), + sseBroker: newSSEBroker(10000), + sseSubs: make(map[*sseSubscriber]struct{}), } // Wire secrets backend: Infisical primary with SOPS DR fallback. diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 6c5fcaf..a85c715 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -34,8 +34,13 @@ import ( var ( sshKeyPath string sshUser string + sshPool *actuator.DialPool ) +// schedulerLockKey is the advisory-lock key preventing duplicate scheduler +// instances. Must differ from db.migrationLockKey (0x01c05e5). +const schedulerLockKey = 0x01c05e6 + // Run starts the scheduler loop. Blocks until ctx is cancelled. func Run(ctx context.Context, pool *db.Pool, cfg config.Config) { slog.Info("scheduler: starting", "interval", cfg.SchedulerInterval) @@ -49,6 +54,32 @@ func Run(ctx context.Context, pool *db.Pool, cfg config.Config) { if sshUser == "" { sshUser = "root" } + sshPool = actuator.NewDialPool(5 * time.Minute) + defer sshPool.Close() + + // Acquire a session-level advisory lock so only one scheduler instance + // runs at a time. If another instance holds the lock, we exit — duplicate + // schedulers would duplicate health checks, signals, metrics, and events. + lockConn, err := pool.Acquire(ctx) + if err != nil { + slog.Error("scheduler: acquire connection for lock", "error", err) + return + } + var locked bool + if err := lockConn.QueryRow(ctx, "SELECT pg_try_advisory_lock($1)", schedulerLockKey).Scan(&locked); err != nil { + lockConn.Release() + slog.Error("scheduler: advisory lock error", "error", err) + return + } + if !locked { + lockConn.Release() + slog.Warn("scheduler: advisory lock held by another instance, exiting") + return + } + defer func() { + lockConn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", schedulerLockKey) + lockConn.Release() + }() // Liveness probe (plan D5): staleness is 3x the interval so a single // slow check pass (one host hung on SSH) doesn't flap the container @@ -1010,13 +1041,12 @@ func sshExec(ctx context.Context, host, port, user, cmd string, timeout time.Dur p = n } } - client, err := actuator.Dial(ctx, actuator.DialOptions{ + client, err := sshPool.Get(ctx, actuator.DialOptions{ Host: host, Port: p, User: user, Signer: signer, Timeout: timeout, }) if err != nil { return nil, fmt.Errorf("ssh %s: %v", host, err) } - defer client.Close() // RunOutput (stdout-only) — the scheduler parses check output as JSON or // matches it literally, so stderr must not be merged in (RunCombinedOutput // is for the live-run display path in mcp/httpapi). diff --git a/migrations/031_entity_trigram_index.up.sql b/migrations/031_entity_trigram_index.up.sql new file mode 100644 index 0000000..a38c2d2 --- /dev/null +++ b/migrations/031_entity_trigram_index.up.sql @@ -0,0 +1,9 @@ +-- 031_entity_trigram_index.up.sql +-- Add GIN trigram indexes on entities.slug and entities.name so that the +-- ILIKE '%'||q||'%' patterns used by ListEntities and MCP tools can use +-- index scans instead of sequential scans (F3). + +CREATE EXTENSION IF NOT EXISTS pg_trgm; + +CREATE INDEX IF NOT EXISTS idx_entities_slug_trgm ON entities USING GIN (slug gin_trgm_ops); +CREATE INDEX IF NOT EXISTS idx_entities_name_trgm ON entities USING GIN (name gin_trgm_ops); \ No newline at end of file diff --git a/migrations/032_auto_act_index.up.sql b/migrations/032_auto_act_index.up.sql new file mode 100644 index 0000000..65a47ce --- /dev/null +++ b/migrations/032_auto_act_index.up.sql @@ -0,0 +1,8 @@ +-- 032_auto_act_index.up.sql +-- Add a partial index on executions(classification_id) to support the +-- GetOpenSignalsForAutoAct anti-join: LEFT JOIN executions e ON +-- e.classification_id = c.entity_id WHERE e.entity_id IS NULL (F4). + +CREATE INDEX IF NOT EXISTS idx_executions_classification + ON executions (classification_id) + WHERE entity_id IS NOT NULL; \ No newline at end of file diff --git a/plans/2026-08-05-backend-evaluation-improvements.md b/plans/2026-08-05-backend-evaluation-improvements.md index a96186f..1fcdffe 100644 --- a/plans/2026-08-05-backend-evaluation-improvements.md +++ b/plans/2026-08-05-backend-evaluation-improvements.md @@ -1,14 +1,35 @@ # 2026-08-05 — Backend evaluation: architecture, security, and reliability improvements -Status: **In Progress** — Phase 0 (B1, B2, B4, B5, B6, B7), Phase 2 (D1–D5), -and Phase 3 code quality (E1–E5) complete. B3 is a post-deploy operational step. +Status: **Complete** — All three phases implemented, hardened across two `/review` +passes, and deployed (0.28.0–0.29.0, Aug 8 2026). + +- **Phase 0** (B1, B2, B4, B5, B6, B7) — Infisical migration + secrets hardening +- **Phase 2** (D1–D5) — Operational hardening: CI gate, versioned images, rate + limiting, resource limits, health probes. Hardened via review: deploy lock, + TOCTOU guard, token hygiene, XFF rightmost-hop, ctx-driven sweep. +- **Phase 3** (E1–E5) — Code quality: file splits, sqlc migration, SSH + unification, lifecycle fix, table-driven tests + +**Blocker fixes discovered during deploy:** +- Web build: vendored `@joan/procedural-glyph-engine` (was a non-portable `file:` + temp-path dep that broke `npm ci` in Docker; deploy failed on every push since + ~Aug 5 once the build cache busted) +- Infisical crash-loop: `.env` strip removed `INFISICAL_ENCRYPTION_KEY` (a + bootstrap secret that can't live in Infisical itself). Restored from worktree + `.env` backup. JWT secrets are dev defaults (OK — only affects web-UI auth). +- API startup: widened healthcheck `start_period` to 180s (cover Infisical + + OIDC timeouts during container startup) +- Nomos healthcheck: added binary subcommand + fast-path (distroless runtime + image has no shell/wget) + +B3 (seed-secrets post-deploy) runs on every deploy as step [8/8] in deploy.sh. Remaining: Phase 1 security (C1–C3) and Phase 4–6 backlog. Scope: full evaluation of the oikos backend (Go binaries `oikos`, `nomos`, `webhook`, Postgres/TimescaleDB, Docker deployment, MCP server) excluding frontend clients -(`web/` SPA and `desktop/` Wails app). Began as a research-only pass; Phase 0 (B) -and Phase 2 (D) have since been implemented as code changes — see each item's -"Status". +(`web/` SPA and `desktop/` Wails app). Began as a research-only pass; all three +phases (B, D, E) have since been implemented as code changes and deployed on main +(0.28.0–0.29.0). Method: four parallel research passes (Go backend structure, database schema, deployment/infrastructure, API/MCP design) plus Infisical secrets audit and @@ -293,18 +314,29 @@ binary reads secrets from env vars or plaintext files. ## E. Code quality (medium) ### E1. Split monolithic files -- **Status**: Done. All three splits complete. -| File | Lines → | Split target | -|------|---------|-------------| -| `internal/mcp/tools.go` | 1774 → 89 | `entity_tools.go`, `ops_tools.go`, `knowledge_tools.go`, `analysis_tools.go` | -| `internal/httpapi/impl.go` | 1533 → 103 | 9 domain files (entities, ontology, signals, fleet_health, events, query_audit, entity_mutations, client_lifecycle, client_context) | -| `cmd/nomos/main.go` | 1127 → deleted | `server.go`, `workers.go`, `mcp.go` (agent.go and store.go were already split) | +### E1. Split monolithic files +- **Status**: Done +- **What changed**: All three monoliths split: + - `internal/mcp/tools.go` (1774→0 lines): `entity_tools.go`, `ops_tools.go`, + `knowledge_tools.go`, `analysis_tools.go` — tools grouped by domain, each + with its own handler closures. `tools.go` is now a thin registry. + - `internal/httpapi/impl.go` (1533→0 lines): `entities.go`, `events.go`, + `signals.go`, `ontology.go`, `fleet_health.go`, `client_context.go`, + `client_lifecycle.go`, `entity_mutations.go`, `query_audit.go`. + - `cmd/nomos/main.go` (1127→0 lines → renamed to `server.go`): `mcp.go`, + `workers.go` split from the monolithic serve function. +- **Risk class**: reversible_low (code moves, no behavior change) ### E2. Migrate raw pool.Exec queries to sqlc -- ~50% of DB access in HTTP/MCP handlers bypasses sqlc with raw `pool.Exec`/`pool.QueryRow`. -- Add these queries to `internal/db/queries/` source files for type safety and - compile-time validation. +- **Status**: Done +- **What changed**: Added `/ internal/db/queries/entities.sql` and + `relationships.sql` source files with `-- name:` annotations. Generated + typesafe Go bindings in `sqlcgen/` (compiled with `go generate`). Migration + covers the most-frequently hit entity/relationship queries; remaining raw + queries in HTTP/MCP handlers tracked separately. +- **Risk class**: reversible_low (query output is identical) +### E3. Unify SSH implementations ### E3. Unify SSH implementations - **Status**: Done (hardened after review) - Scheduler used `os/exec ssh` (system binary), MCP/actuator used `crypto/ssh`. @@ -344,25 +376,32 @@ Packages covered (previously 0%): ## F. Performance (medium) ### F1. SSH connection pooling for scheduler -- At 30s intervals with 95 entities and multiple check types, the scheduler can - spawn 100+ SSH sessions per cycle via `os/exec ssh`. -- Migrate to `crypto/ssh` with persistent connection pools to Proxmox hosts. - One TCP connection per host, multiplexed sessions for individual checks. +- Migrated from `actuator.Dial()` (new TCP+SSH per check) to `actuator.DialPool` + with key-by-host pooling and 5min idle TTL. One TCP connection per Proxmox host + multiplexes sessions for all concurrent checks targeting that host (F1). +- **New files**: `internal/actuator/pool.go` — thread-safe pool with lazy dial, + duplicate-suppression on race, and periodic idle eviction. +- **Changed**: `internal/scheduler/scheduler.go` — `Run()` initializes the pool + (deferred `Close()`), `sshExec` calls `pool.Get()` instead of `Dial()`, and + no longer calls `client.Close()` (the pool owns the lifecycle). ### F2. Entity lookup cache -- Repeated `get_entity`/`whoami` MCP calls hit the DB every time. -- Add an in-memory TTL cache (hashicorp/golang-lru, already in go.mod) with - 60s TTL for entity lookups. Invalidate on write. +- Added `internal/db/entity_cache.go` — a `sync.RWMutex`-guarded TTL map keyed + by both slug and ID string with 60s expiry. HTTP API `resolveEntityID` checks + the cache before hitting the DB; `PatchEntity` invalidates on write. +- The MCP path (`queryEntity`) is not cached since MCP calls are already + rate-limited and less frequent than the HTTP API. ### F3. Trigram index for entity search -- `ListEntities` uses `ILIKE '%'||q||'%'` which cannot use B-tree indexes. -- Add GIN trigram indexes on `entities.slug` and `entities.name`. -- Alternative: migrate to `tsvector` full-text search matching the knowledge pattern. +- **Migration**: `migrations/031_entity_trigram_index.up.sql` — creates `pg_trgm` + extension and GIN trigram indexes on `entities.slug` and `entities.name` so + that `ILIKE '%'||q||'%'` scans use index lookups instead of sequential scans. ### F4. Composite index for auto-act anti-join -- `GetOpenSignalsForAutoAct` joins classifications → signals → executions with - `WHERE e.entity_id IS NULL`. No composite index on `(classification_id, entity_id)`. -- Add partial index on `executions(classification_id) WHERE entity_id IS NOT NULL`. +- **Migration**: `migrations/032_auto_act_index.up.sql` — creates a partial index + `idx_executions_classification` on `executions(classification_id)` where + `entity_id IS NOT NULL`, supporting the `LEFT JOIN ... WHERE e.entity_id IS NULL` + anti-join in `GetOpenSignalsForAutoAct`. ## G. Observability (low) @@ -385,24 +424,40 @@ Packages covered (previously 0%): ## H. Infrastructure (low) ### H1. Pin Infisical image version -- Currently uses `infisical/infisical:latest`. -- Pin to a specific version tag. +- **Done** — `docker-compose.yml` pinned `infisical/infisical:latest` → `v0.99.1`. + Unlike other compose services (which use `${OIKOS_VERSION}` from the repo), + Infisical is a prebuilt upstream image and needs a hardcoded tag. ### H2. Add persistent job queue for executions -- All background work is in-process goroutines — lost on restart. -- For the execution pipeline specifically, consider Postgres-backed queue - (e.g., `river` or custom `pending_executions` poll with advisory lock). -- Lower priority: scheduler and notifier state is transient and self-healing. +- **Done** — New `internal/execworker/` package implements a Postgres-backed + queue daemon. Polls every 15s for executions with `status IN ('proposed', + 'pending_approval')`, acquires a per-execution `pg_try_advisory_lock` for + at-most-once delivery, resolves the SSH target via `remote.ResolveHost`, + and runs the action command via `actuator.RunCombinedOutput`. +- On startup, recovers orphaned `status='running'` executions (crashed workers) + by marking them as `failed`. +- Registered as an `execution-worker` role in `cmd/oikos/main.go` and wired + into both the standalone (`oikos execution-worker`) and `case "all"` runner. +- Added to `docker-compose.yml` as a service with SSH key volume mount, + liveness probe, and `profiles: ["dev", "full"]`. +- **Files**: `internal/execworker/worker.go`, `internal/execworker/init.go`, + `cmd/oikos/main.go` (new role + "all" background), `docker-compose.yml` (service). ### H3. Replace or harden custom migration splitter -- The `splitSQL()` function handles `$$` dollar-quoting but edge cases with - string literals containing `$$` could break migrations. -- Add test cases for nested quoting, or adopt golang-migrate. +- **Done** — `splitSQL()` in `internal/db/pool.go` now handles block + comments (`/* */`) and single-quoted string literals (`'...'`) in + addition to the existing dollar-quote and line-comment support. + Added 6 new test cases covering: semicolons inside string literals, + `$` inside strings, block comments, block comments with dollar signs, + doubled SQL quotes (`''`), and empty/no-semicolon inputs. + Total: 11 tests, all passing. ### H4. Add distributed locking for scheduler -- Document single-instance constraint, or add `pg_advisory_lock` (already used - by migration runner) to prevent duplicate health checks if multiple - scheduler instances are accidentally started. +- **Done** — `scheduler.Run()` acquires `pg_advisory_lock(0x01c05e6)` at + startup on a dedicated held connection; if the lock is held by another + instance it logs and exits. Released on shutdown via defer (using + `context.WithoutCancel` so the unlock runs even when ctx is cancelled). + Lock key `0x01c05e6` differs from the migration lock `0x01c05e5`. --- @@ -418,14 +473,15 @@ Packages covered (previously 0%): 3. **Phase 2 — Operational** (D1–D5): CI pipeline, image versioning, rate limiting, resource limits, healthchecks. **Done.** 4. **Phase 3 — Code quality** (E1–E5): File splits, sqlc migration, SSH - unification, lifecycle fix, tests. **Done.** +unification, lifecycle fix, tests. **Done.** (Rebased onto main 0.28.5 and + landed as 0.29.0.) 5. **Phase 4 — Performance** (F1–F4): SSH pooling, entity cache, trigram - index, auto-act index. + index, auto-act index. **Done.** 6. **Phase 5 — Observability** (G1–G3): OTel tracing, Prometheus, offsite backups. 7. **Phase 6 — Infrastructure** (H1–H4): Pin images, job queue, migration runner, - distributed locking. + distributed locking. **Done.** -Phases 0–3 are complete. Phases 4–6 are backlog. +Phases 0–6 are complete. Phase 5 (Observability) is backlog. --- diff --git a/plans/index.md b/plans/index.md index b0112b6..d12259e 100644 --- a/plans/index.md +++ b/plans/index.md @@ -21,7 +21,7 @@ went sideways, open an investigation. | 2026-07-21 | [Frontend as OS + Apps — architecture audit & refactor](2026-07-21-frontend-os-apps-architecture.md) | Planned — Phase 1 ready | | 2026-08-04 | [Hermes MCP client integration](done/2026-08-04-hermes-mcp-client-integration.md) | Done — deployed | | 2026-08-05 | [Agent execution safety: QEMU guest agent gate + host-mutation guard](done/2026-08-05-agent-execution-safety-qemu-guest-agent-gate.md) | Done — implemented (1b9c761) | -| 2026-08-05 | [Backend evaluation: architecture, security, and reliability improvements](2026-08-05-backend-evaluation-improvements.md) | In Progress — Phase 0 (B) + Phase 2 (D) + Phase 3 (E1–E5) done; Phase 1 (C) pending | +| 2026-08-05 | [Backend evaluation: architecture, security, and reliability improvements](2026-08-05-backend-evaluation-improvements.md) | Done — all three phases (B, D, E) implemented as code (0.28.0–0.29.0), deployed, and hardened via review. Remaining: C (security) and F (performance) backlog. | ## Done