Simplify plan storage and HTTP handling; cache compiled schemas

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)
 }

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)
 }

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

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
+}

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)
+			}
+		}
+	})
+}