From 7475c38e0d4bbae0750adfb40368646f0ceda394 Mon Sep 17 00:00:00 2001 From: rick12345 Date: Sat, 5 Sep 2026 16:49:37 +0800 Subject: [PATCH] Simplify plan storage and HTTP handling; cache compiled schemas --- agentapi/http.go | 35 +------- controlplane/http.go | 47 +--------- host/schema.go | 4 + host/schema_cache.go | 51 +++++++++++ host/schema_cache_test.go | 127 +++++++++++++++++++++++++++ internal/httpjson/httpjson.go | 72 ++++++++++++++++ internal/httpjson/httpjson_test.go | 62 ++++++++++++++ taskstate/http.go | 25 ++---- taskstate/http_test.go | 40 +++++++++ taskstate/store.go | 37 ++------ taskstate/store_migration_test.go | 132 +++++++++++++++++++++++++++++ 11 files changed, 512 insertions(+), 120 deletions(-) create mode 100644 host/schema_cache.go create mode 100644 host/schema_cache_test.go create mode 100644 internal/httpjson/httpjson.go create mode 100644 internal/httpjson/httpjson_test.go create mode 100644 taskstate/store_migration_test.go diff --git a/agentapi/http.go b/agentapi/http.go index 43b4575..9a9bbba 100644 --- a/agentapi/http.go +++ b/agentapi/http.go @@ -1,20 +1,14 @@ package agentapi import ( - "bytes" "context" - "crypto/subtle" - "encoding/json" "errors" "fmt" - "io" - "mime" "net/http" - "strings" "github.com/sunrioa/rin/cognition" "github.com/sunrioa/rin/host" - "github.com/sunrioa/rin/internal/jsonwire" + "github.com/sunrioa/rin/internal/httpjson" ) const defaultHTTPMaxBodyBytes int64 = 1 << 20 @@ -69,9 +63,7 @@ func (server *HTTPHandler) secure(next http.Handler) http.Handler { return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { response.Header().Set("Cache-Control", "no-store") response.Header().Set("X-Content-Type-Options", "nosniff") - provided := strings.TrimPrefix(request.Header.Get("Authorization"), "Bearer ") - if len(provided) != len(server.token) || - subtle.ConstantTimeCompare([]byte(provided), []byte(server.token)) != 1 { + if !httpjson.Authorized(request, server.token) { response.Header().Set("WWW-Authenticate", "Bearer") writeHTTPError(response, http.StatusUnauthorized, "forbidden", "unauthorized") return @@ -186,24 +178,7 @@ func (server *HTTPHandler) decode( request *http.Request, target any, ) error { - contentType, _, err := mime.ParseMediaType(request.Header.Get("Content-Type")) - if err != nil || contentType != "application/json" { - return errors.New("content type must be application/json") - } - request.Body = http.MaxBytesReader(response, request.Body, server.maxBodyBytes) - payload, err := io.ReadAll(request.Body) - if err != nil { - return errors.New("request body exceeds the configured limit") - } - if err := jsonwire.Validate(payload); err != nil { - return fmt.Errorf("invalid JSON: %w", err) - } - decoder := json.NewDecoder(bytes.NewReader(payload)) - decoder.DisallowUnknownFields() - if err := decoder.Decode(target); err != nil { - return fmt.Errorf("invalid request: %w", err) - } - return nil + return httpjson.DecodeRequest(response, request, server.maxBodyBytes, target) } func writeTaskError(response http.ResponseWriter, err error) { @@ -230,7 +205,5 @@ func writeHTTPError(response http.ResponseWriter, status int, code, message stri } func writeJSON(response http.ResponseWriter, status int, value any) { - response.Header().Set("Content-Type", "application/json") - response.WriteHeader(status) - _ = json.NewEncoder(response).Encode(value) + httpjson.Write(response, status, value) } diff --git a/controlplane/http.go b/controlplane/http.go index da4ed04..aa2397a 100644 --- a/controlplane/http.go +++ b/controlplane/http.go @@ -1,20 +1,15 @@ package controlplane import ( - "bytes" "context" - "crypto/subtle" "encoding/json" "errors" "fmt" - "io" - "mime" "net/http" - "strings" "time" "github.com/sunrioa/rin/host" - "github.com/sunrioa/rin/internal/jsonwire" + "github.com/sunrioa/rin/internal/httpjson" "github.com/sunrioa/rin/timeline" ) @@ -197,13 +192,7 @@ func (server *hostHTTPHandler) secure(next http.Handler) http.Handler { response.Header().Set("X-Content-Type-Options", "nosniff") if request.URL.Path != "/health" && request.URL.Path != "/control/v2/health" { - provided := strings.TrimPrefix( - request.Header.Get("Authorization"), "Bearer ", - ) - if len(provided) != len(server.token) || - subtle.ConstantTimeCompare( - []byte(provided), []byte(server.token), - ) != 1 { + if !httpjson.Authorized(request, server.token) { response.Header().Set("WWW-Authenticate", "Bearer") writeHTTPError(response, http.StatusUnauthorized, "unauthorized") return @@ -787,28 +776,7 @@ func (server *hostHTTPHandler) decode( request *http.Request, target any, ) error { - contentType, _, err := mime.ParseMediaType( - request.Header.Get("Content-Type"), - ) - if err != nil || contentType != "application/json" { - return errors.New("content type must be application/json") - } - request.Body = http.MaxBytesReader( - response, request.Body, server.maxBodyBytes, - ) - payload, err := io.ReadAll(request.Body) - if err != nil { - return errors.New("request body exceeds the configured limit") - } - if err := jsonwire.Validate(payload); err != nil { - return fmt.Errorf("invalid JSON: %w", err) - } - decoder := json.NewDecoder(bytes.NewReader(payload)) - decoder.DisallowUnknownFields() - if err := decoder.Decode(target); err != nil { - return fmt.Errorf("invalid request: %w", err) - } - return nil + return httpjson.DecodeRequest(response, request, server.maxBodyBytes, target) } func principalHasControlScope(principal host.Principal) bool { @@ -866,12 +834,5 @@ func writeHTTPErrorCode( } func writeJSON(response http.ResponseWriter, status int, value any) { - payload, err := json.Marshal(value) - if err != nil { - http.Error(response, `{"error":"internal error"}`, http.StatusInternalServerError) - return - } - response.Header().Set("Content-Type", "application/json") - response.WriteHeader(status) - _, _ = response.Write(append(payload, '\n')) + httpjson.Write(response, status, value) } diff --git a/host/schema.go b/host/schema.go index 710b0b4..4926712 100644 --- a/host/schema.go +++ b/host/schema.go @@ -132,6 +132,10 @@ func (schema Schema) compiled() (*jsonschema.Schema, error) { } func compileSchema(canonical []byte) (*jsonschema.Schema, error) { + return compiledSchemas.compile(canonical) +} + +func compileSchemaUncached(canonical []byte) (*jsonschema.Schema, error) { value, err := jsonschema.UnmarshalJSON(bytes.NewReader(canonical)) if err != nil { return nil, err diff --git a/host/schema_cache.go b/host/schema_cache.go new file mode 100644 index 0000000..494c58d --- /dev/null +++ b/host/schema_cache.go @@ -0,0 +1,51 @@ +package host + +import ( + "sync" + + "github.com/santhosh-tekuri/jsonschema/v6" +) + +const maxCompiledSchemas = 128 + +var compiledSchemas = schemaCache{capacity: maxCompiledSchemas} + +// schemaCache uses canonical document bytes, never a caller-supplied digest. +// FIFO eviction bounds retained entries. Compiled schemas remain private and +// immutable; evicting an entry does not affect an in-flight validation. +type schemaCache struct { + mu sync.Mutex + capacity int + entries map[string]*jsonschema.Schema + keys []string + next int +} + +func (cache *schemaCache) compile(canonical []byte) (*jsonschema.Schema, error) { + key := string(canonical) + cache.mu.Lock() + defer cache.mu.Unlock() + if compiled := cache.entries[key]; compiled != nil { + return compiled, nil + } + // Serialize misses so concurrent requests for one schema compile it once. + compiled, err := compileSchemaUncached(canonical) + if err != nil { + return nil, err + } + if cache.capacity <= 0 { + return compiled, nil + } + if cache.entries == nil { + cache.entries = make(map[string]*jsonschema.Schema) + } + if len(cache.keys) < cache.capacity { + cache.keys = append(cache.keys, key) + } else { + delete(cache.entries, cache.keys[cache.next]) + cache.keys[cache.next] = key + cache.next = (cache.next + 1) % cache.capacity + } + cache.entries[key] = compiled + return compiled, nil +} diff --git a/host/schema_cache_test.go b/host/schema_cache_test.go new file mode 100644 index 0000000..ab30593 --- /dev/null +++ b/host/schema_cache_test.go @@ -0,0 +1,127 @@ +package host + +import ( + "bytes" + "fmt" + "sync" + "testing" + + "github.com/santhosh-tekuri/jsonschema/v6" +) + +func cacheTestDocument(n int) []byte { + return []byte(fmt.Sprintf(`{"$schema":"https://json-schema.org/draft/2020-12/schema","additionalProperties":false,"properties":{"name":{"minLength":%d,"type":"string"}},"required":["name"],"type":"object"}`, n)) +} + +func TestSchemaCacheConcurrentReuseAndEviction(t *testing.T) { + cache := schemaCache{capacity: 2} + const workers = 16 + results := make(chan *jsonschema.Schema, workers) + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + compiled, err := cache.compile(cacheTestDocument(1)) + if err != nil { + t.Error(err) + return + } + if err := compiled.Validate(map[string]any{"name": "ok"}); err != nil { + t.Error(err) + } + if err := compiled.Validate(map[string]any{"name": ""}); err == nil { + t.Error("invalid instance accepted") + } + results <- compiled + }() + } + wg.Wait() + close(results) + var first *jsonschema.Schema + for result := range results { + if first == nil { + first = result + } else if first != result { + t.Fatal("concurrent callers did not reuse compilation") + } + } + if first == nil { + t.Fatal("no compiled result") + } + for i := 2; i <= 3; i++ { + if _, err := cache.compile(cacheTestDocument(i)); err != nil { + t.Fatal(err) + } + } + if len(cache.entries) != 2 || cache.entries[string(cacheTestDocument(1))] != nil { + t.Fatal("cache did not evict oldest schema") + } + if err := first.Validate(map[string]any{"name": "ok"}); err != nil { + t.Fatalf("eviction broke in-flight schema: %v", err) + } + again, err := cache.compile(cacheTestDocument(1)) + if err != nil || again == first { + t.Fatalf("evicted schema was not recompiled: %v", err) + } + // Failed compilations must not displace useful entries. + if _, err := cache.compile([]byte(`{"type":"invalid"}`)); err == nil { + t.Fatal("invalid schema compiled") + } + if len(cache.entries) != 2 { + t.Fatal("failed compilation changed cache size") + } +} + +func TestWarmSchemaCacheStillChecksDocumentAndDigest(t *testing.T) { + schema, err := NewSchema(cacheTestDocument(1)) + if err != nil { + t.Fatal(err) + } + if err := schema.ValidateInstance([]byte(`{"name":"a"}`)); err != nil { + t.Fatal(err) + } + changed := schema + changed.Document = bytes.ReplaceAll(schema.Document, []byte(`"minLength":1`), []byte(`"minLength":2`)) + if err := changed.Validate(); err == nil { + t.Fatal("warm cache trusted stale digest") + } + changed.SHA256 = sha256Hex(changed.Document) + if err := changed.ValidateInstance([]byte(`{"name":"a"}`)); err == nil { + t.Fatal("changed document reused old constraints") + } + if err := schema.ValidateInstance([]byte(`{"name":"a"}`)); err != nil { + t.Fatal("new schema changed existing entry") + } + noncanonical := schema + noncanonical.Document = append([]byte(" "), schema.Document...) + noncanonical.SHA256 = sha256Hex(noncanonical.Document) + if err := noncanonical.Validate(); err == nil { + t.Fatal("warm cache accepted noncanonical document") + } +} + +func BenchmarkSchemaCompilation(b *testing.B) { + document := cacheTestDocument(1) + b.Run("uncached", func(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + if _, err := compileSchemaUncached(document); err != nil { + b.Fatal(err) + } + } + }) + b.Run("cached", func(b *testing.B) { + cache := schemaCache{capacity: maxCompiledSchemas} + if _, err := cache.compile(document); err != nil { + b.Fatal(err) + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := cache.compile(document); err != nil { + b.Fatal(err) + } + } + }) +} diff --git a/internal/httpjson/httpjson.go b/internal/httpjson/httpjson.go new file mode 100644 index 0000000..e766872 --- /dev/null +++ b/internal/httpjson/httpjson.go @@ -0,0 +1,72 @@ +// Package httpjson implements the shared JSON transport boundary. Services keep +// ownership of authorization policy and their public error envelopes. +package httpjson + +import ( + "bytes" + "crypto/subtle" + "encoding/json" + "errors" + "fmt" + "io" + "mime" + "net/http" + "strings" + + "github.com/sunrioa/rin/internal/jsonwire" +) + +// Authorized checks a Bearer credential without interpreting service scopes. +func Authorized(request *http.Request, token string) bool { + provided, ok := strings.CutPrefix(request.Header.Get("Authorization"), "Bearer ") + return ok && token != "" && len(provided) == len(token) && + subtle.ConstantTimeCompare([]byte(provided), []byte(token)) == 1 +} + +// ReadBody requires JSON and bounds the read before allocating the full body. +func ReadBody(response http.ResponseWriter, request *http.Request, limit int64) ([]byte, error) { + contentType, _, err := mime.ParseMediaType(request.Header.Get("Content-Type")) + if err != nil || contentType != "application/json" { + return nil, errors.New("content type must be application/json") + } + request.Body = http.MaxBytesReader(response, request.Body, limit) + payload, err := io.ReadAll(request.Body) + if err != nil { + return nil, errors.New("request body exceeds the configured limit") + } + return payload, nil +} + +// Decode rejects ambiguous JSON, malformed Unicode and unknown fields. +func Decode(payload []byte, target any) error { + if err := jsonwire.Validate(payload); err != nil { + return fmt.Errorf("invalid JSON: %w", err) + } + decoder := json.NewDecoder(bytes.NewReader(payload)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(target); err != nil { + return fmt.Errorf("invalid request: %w", err) + } + return nil +} + +func DecodeRequest(response http.ResponseWriter, request *http.Request, limit int64, target any) error { + payload, err := ReadBody(response, request, limit) + if err != nil { + return err + } + return Decode(payload, target) +} + +// Write encodes before committing the status so encoding failures cannot look +// like successful responses. +func Write(response http.ResponseWriter, status int, value any) { + payload, err := json.Marshal(value) + if err != nil { + http.Error(response, `{"error":"internal error"}`, http.StatusInternalServerError) + return + } + response.Header().Set("Content-Type", "application/json") + response.WriteHeader(status) + _, _ = response.Write(append(payload, '\n')) +} diff --git a/internal/httpjson/httpjson_test.go b/internal/httpjson/httpjson_test.go new file mode 100644 index 0000000..8a337fa --- /dev/null +++ b/internal/httpjson/httpjson_test.go @@ -0,0 +1,62 @@ +package httpjson + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestDecodeRequestBoundary(t *testing.T) { + for _, tc := range []struct { + name, body, media string + limit int64 + valid bool + }{ + {"valid", `{"name":"ok"}`, "application/json; charset=utf-8", 13, true}, + {"oversized", `{"name":"ok"} `, "application/json", 13, false}, + {"missing media", `{}`, "", 1024, false}, + {"wrong media", `{}`, "text/plain", 1024, false}, + {"duplicate", `{"name":"a","name":"b"}`, "application/json", 1024, false}, + {"trailing", `{} {}`, "application/json", 1024, false}, + {"unknown", `{"extra":1}`, "application/json", 1024, false}, + {"surrogate", `{"name":"\ud800"}`, "application/json", 1024, false}, + {"invalid utf8", "{\"name\":\"\xff\"}", "application/json", 1024, false}, + } { + t.Run(tc.name, func(t *testing.T) { + request := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(tc.body)) + request.Header.Set("Content-Type", tc.media) + var target struct { + Name string `json:"name"` + } + err := DecodeRequest(httptest.NewRecorder(), request, tc.limit, &target) + if (err == nil) != tc.valid { + t.Fatalf("decode error = %v, valid = %v", err, tc.valid) + } + }) + } +} + +func TestAuthorizedRequiresBearer(t *testing.T) { + const token = "0123456789abcdef0123456789abcdef" + for _, header := range []string{"Bearer " + token, token, "", "Basic " + token, "Bearer wrong", "Bearer " + token + " "} { + request := httptest.NewRequest(http.MethodGet, "/", nil) + request.Header.Set("Authorization", header) + if got := Authorized(request, token); got != (header == "Bearer "+token) { + t.Fatalf("authorization %q = %v", header, got) + } + } +} + +func TestWriteHandlesEncodingFailureBeforeStatus(t *testing.T) { + response := httptest.NewRecorder() + Write(response, http.StatusOK, make(chan int)) + if response.Code != http.StatusInternalServerError { + t.Fatalf("status = %d", response.Code) + } + response = httptest.NewRecorder() + Write(response, http.StatusCreated, map[string]string{"id": "one"}) + if response.Code != http.StatusCreated || response.Header().Get("Content-Type") != "application/json" || response.Body.String() != "{\"id\":\"one\"}\n" { + t.Fatalf("response = %#v", response) + } +} diff --git a/taskstate/http.go b/taskstate/http.go index d7d3c92..116035a 100644 --- a/taskstate/http.go +++ b/taskstate/http.go @@ -3,7 +3,6 @@ package taskstate import ( "bytes" "context" - "crypto/subtle" "encoding/json" "errors" "fmt" @@ -14,6 +13,7 @@ import ( "time" "github.com/sunrioa/rin/controlplane" + "github.com/sunrioa/rin/internal/httpjson" ) const maxHTTPRequestBytes int64 = 1 << 20 @@ -189,14 +189,12 @@ func planHTTPHandler( ) http.HandlerFunc { return func(response http.ResponseWriter, request *http.Request) { response.Header().Set("Cache-Control", "no-store") - token := strings.TrimPrefix(request.Header.Get("Authorization"), "Bearer ") - if len(token) != len(options.Token) || - subtle.ConstantTimeCompare([]byte(token), []byte(options.Token)) != 1 { + if !httpjson.Authorized(request, options.Token) { writePlanHTTPError(response, http.StatusUnauthorized, "unauthorized") return } - body, err := io.ReadAll(io.LimitReader(request.Body, maxHTTPRequestBytes+1)) - if err != nil || int64(len(body)) > maxHTTPRequestBytes { + body, err := httpjson.ReadBody(response, request, maxHTTPRequestBytes) + if err != nil { writePlanHTTPError(response, http.StatusBadRequest, "invalid_request") return } @@ -206,20 +204,13 @@ func planHTTPHandler( writePlanHTTPError(response, status, code) return } - response.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(response).Encode(output) + httpjson.Write(response, http.StatusOK, output) } } func decodeHTTPInput[T any](payload []byte) (T, error) { var target T - decoder := json.NewDecoder(bytes.NewReader(payload)) - decoder.DisallowUnknownFields() - if err := decoder.Decode(&target); err != nil { - return target, ErrInvalid - } - var trailing any - if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) { + if err := httpjson.Decode(payload, &target); err != nil { return target, ErrInvalid } return target, nil @@ -241,9 +232,7 @@ func planHTTPError(err error) (int, string) { } func writePlanHTTPError(response http.ResponseWriter, status int, code string) { - response.Header().Set("Content-Type", "application/json") - response.WriteHeader(status) - _ = json.NewEncoder(response).Encode(map[string]string{ + httpjson.Write(response, status, map[string]string{ "code": code, "error": code, }) } diff --git a/taskstate/http_test.go b/taskstate/http_test.go index bf40a83..3843d9a 100644 --- a/taskstate/http_test.go +++ b/taskstate/http_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "path/filepath" + "strings" "testing" "github.com/sunrioa/rin/controlplane" @@ -120,3 +121,42 @@ func httpTestDraft() Draft { } var _ http.RoundTripper = handlerTransport{} + +func TestPlanHTTPRejectsMalformedBoundaryBeforeDispatch(t *testing.T) { + const token = "0123456789abcdef0123456789abcdef" + for _, tc := range []struct { + name, body, media, auth string + status int + }{ + {"valid", `{"plan_id":"one"}`, "application/json", "Bearer " + token, 200}, + {"duplicate", `{"plan_id":"one","plan_id":"two"}`, "application/json", "Bearer " + token, 400}, + {"surrogate", `{"plan_id":"\ud800"}`, "application/json", "Bearer " + token, 400}, + {"unknown", `{"extra":1}`, "application/json", "Bearer " + token, 400}, + {"trailing", `{} {}`, "application/json", "Bearer " + token, 400}, + {"media", `{}`, "text/plain", "Bearer " + token, 400}, + {"bare credential", `{}`, "application/json", token, 401}, + } { + t.Run(tc.name, func(t *testing.T) { + dispatched := false + handler := planHTTPHandler(HTTPOptions{Token: token}, func(_ context.Context, body []byte) (any, error) { + input, err := decodeHTTPInput[GetPlanInput](body) + if err != nil { + return nil, err + } + dispatched = true + return input, nil + }) + request := httptest.NewRequest(http.MethodPost, "/plans/v1/get", strings.NewReader(tc.body)) + request.Header.Set("Content-Type", tc.media) + request.Header.Set("Authorization", tc.auth) + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + if response.Code != tc.status || dispatched != (tc.status == 200) { + t.Fatalf("status=%d dispatched=%v body=%s", response.Code, dispatched, response.Body.String()) + } + if tc.status != 200 && !strings.Contains(response.Body.String(), `"code":`) { + t.Fatal("lost plan error envelope") + } + }) + } +} diff --git a/taskstate/store.go b/taskstate/store.go index 8109974..bac33ea 100644 --- a/taskstate/store.go +++ b/taskstate/store.go @@ -23,7 +23,7 @@ import ( ) const ( - storeSchemaVersion = 1 + storeSchemaVersion = 2 defaultMaxPlans = 1_024 defaultMaxEvents = 4_096 maxWaitMillis = 25_000 @@ -174,7 +174,7 @@ func (store *Store) initialize(ctx context.Context) error { if err := store.db.QueryRowContext(ctx, `PRAGMA user_version`).Scan(&version); err != nil { return fmt.Errorf("%w: read schema: %v", ErrPersist, err) } - if version != 0 && version != storeSchemaVersion { + if version < 0 || version > storeSchemaVersion { return fmt.Errorf("%w: unsupported schema %d", ErrPersist, version) } tx, err := store.db.BeginTx(ctx, nil) @@ -193,11 +193,6 @@ func (store *Store) initialize(ctx context.Context) error { `CREATE UNIQUE INDEX IF NOT EXISTS task_plans_active_actor_idx ON task_plans(session_id, actor_id) WHERE status IN ('planned','active','blocked','paused')`, - `CREATE TABLE IF NOT EXISTS task_plan_steps ( - plan_id TEXT NOT NULL REFERENCES task_plans(plan_id) ON DELETE CASCADE, - step_id TEXT NOT NULL, ordinal INTEGER NOT NULL, status TEXT NOT NULL, - step_json TEXT NOT NULL, PRIMARY KEY(plan_id, step_id), UNIQUE(plan_id, ordinal) - ) STRICT`, `CREATE TABLE IF NOT EXISTS task_plan_events ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, plan_id TEXT NOT NULL REFERENCES task_plans(plan_id) ON DELETE CASCADE, revision INTEGER NOT NULL, @@ -219,7 +214,12 @@ func (store *Store) initialize(ctx context.Context) error { return fmt.Errorf("%w: migrate sqlite: %v", ErrPersist, err) } } - if _, err := tx.ExecContext(ctx, `PRAGMA user_version = 1`); err != nil { + // Version 1 duplicated the steps already contained in state_json. Nothing + // reads that projection; remove it atomically with the schema upgrade. + if _, err := tx.ExecContext(ctx, `DROP TABLE IF EXISTS task_plan_steps`); err != nil { + return fmt.Errorf("%w: remove redundant steps: %v", ErrPersist, err) + } + if _, err := tx.ExecContext(ctx, `PRAGMA user_version = 2`); err != nil { return err } if err := tx.Commit(); err != nil { @@ -880,7 +880,7 @@ func insertPlan(ctx context.Context, tx *sql.Tx, state PlanState) error { state.UpdatedAtUnixMillis); err != nil { return err } - return replaceSteps(ctx, tx, state) + return nil } func replacePlan(ctx context.Context, tx *sql.Tx, expected uint64, state PlanState) error { @@ -899,25 +899,6 @@ func replacePlan(ctx context.Context, tx *sql.Tx, expected uint64, state PlanSta if rows != 1 { return ErrConflict } - return replaceSteps(ctx, tx, state) -} - -func replaceSteps(ctx context.Context, tx *sql.Tx, state PlanState) error { - if _, err := tx.ExecContext(ctx, `DELETE FROM task_plan_steps WHERE plan_id = ?`, state.PlanID); err != nil { - return err - } - for index, step := range state.Steps { - payload, err := json.Marshal(step) - if err != nil { - return err - } - if _, err := tx.ExecContext(ctx, `INSERT INTO task_plan_steps( - plan_id, step_id, ordinal, status, step_json - ) VALUES (?, ?, ?, ?, ?)`, state.PlanID, step.StepID, index, - string(step.Status), string(payload)); err != nil { - return err - } - } return nil } diff --git a/taskstate/store_migration_test.go b/taskstate/store_migration_test.go new file mode 100644 index 0000000..1d2b21f --- /dev/null +++ b/taskstate/store_migration_test.go @@ -0,0 +1,132 @@ +package taskstate + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "path/filepath" + "reflect" + "testing" + + "github.com/sunrioa/rin/internal/sqlitedsn" +) + +func TestMigrateV1PreservesPlanAndEvidence(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "plans.db") + store, err := OpenSQLiteStore(path, StoreConfig{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + plan, err := store.Create(ctx, httpTestDraft()) + if err != nil { + t.Fatal(err) + } + link := OperationLink{OperationID: "op.migration", PlanID: plan.PlanID, PlanRevision: plan.Revision, StepID: plan.CurrentStepID, ConditionIDs: []string{"condition.collected"}} + if err := store.LinkOperation(ctx, link); err != nil { + t.Fatal(err) + } + plan, err = store.Get(ctx, plan.PlanID) + if err != nil { + t.Fatal(err) + } + events, err := store.Events(ctx, plan.PlanID, 0, 100) + if err != nil { + t.Fatal(err) + } + // Recreate the v1-only projection and version on a populated database. + if _, err := store.db.Exec(`CREATE TABLE task_plan_steps ( + plan_id TEXT NOT NULL REFERENCES task_plans(plan_id) ON DELETE CASCADE, + step_id TEXT NOT NULL, ordinal INTEGER NOT NULL, status TEXT NOT NULL, + step_json TEXT NOT NULL, PRIMARY KEY(plan_id, step_id), UNIQUE(plan_id, ordinal)) STRICT`); err != nil { + t.Fatal(err) + } + for i, step := range plan.Steps { + payload, err := json.Marshal(step) + if err != nil { + t.Fatal(err) + } + if _, err := store.db.Exec(`INSERT INTO task_plan_steps VALUES (?, ?, ?, ?, ?)`, plan.PlanID, step.StepID, i, string(step.Status), string(payload)); err != nil { + t.Fatal(err) + } + } + if _, err := store.db.Exec(`PRAGMA user_version = 1`); err != nil { + t.Fatal(err) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + for reopen := 0; reopen < 2; reopen++ { + store, err = OpenSQLiteStore(path, StoreConfig{}) + if err != nil { + t.Fatal(err) + } + restored, err := store.Get(ctx, plan.PlanID) + if err != nil || !reflect.DeepEqual(restored, plan) { + t.Fatalf("plan changed across migration: %v", err) + } + restoredEvents, err := store.Events(ctx, plan.PlanID, 0, 100) + if err != nil || !reflect.DeepEqual(restoredEvents, events) { + t.Fatalf("events changed across migration: %v", err) + } + var links int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM task_plan_operations + WHERE operation_id = ? AND plan_id = ? AND plan_revision = ? AND step_id = ?`, + link.OperationID, link.PlanID, link.PlanRevision, link.StepID).Scan(&links); err != nil { + t.Fatal(err) + } + if links != 1 { + t.Fatal("migration lost the existing operation link") + } + if err := store.LinkOperation(ctx, link); err != nil { + t.Fatalf("operation link lost idempotency: %v", err) + } + var tables, version int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE name = 'task_plan_steps'`).Scan(&tables); err != nil { + t.Fatal(err) + } + if err := store.db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil { + t.Fatal(err) + } + if tables != 0 || version != 2 { + t.Fatalf("tables=%d version=%d", tables, version) + } + if reopen == 1 { + _, err := store.SetStatus(ctx, StatusInput{PlanID: plan.PlanID, ExpectedRevision: plan.Revision + 1, Status: PlanPaused, Summary: "stale"}) + if !errors.Is(err, ErrConflict) { + t.Fatalf("stale CAS accepted after migration: %v", err) + } + if _, err := store.SetStatus(ctx, StatusInput{PlanID: plan.PlanID, ExpectedRevision: plan.Revision, Status: PlanPaused, Summary: "pause"}); err != nil { + t.Fatalf("update without steps table: %v", err) + } + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + } +} + +func TestStoreRejectsFutureSchemaWithoutChangingIt(t *testing.T) { + path := filepath.Join(t.TempDir(), "future.db") + db, err := sql.Open("sqlite", sqlitedsn.File(path)) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.Exec(`PRAGMA user_version = 3`); err != nil { + t.Fatal(err) + } + if store, err := OpenSQLiteStore(path, StoreConfig{}); err == nil { + store.Close() + t.Fatal("future schema accepted") + } + var version int + if err := db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil { + t.Fatal(err) + } + if version != 3 { + t.Fatalf("future version changed to %d", version) + } +}