diff --git a/internal/platform/canvas_model_http.go b/internal/platform/canvas_model_http.go index 89b069b..40800bb 100644 --- a/internal/platform/canvas_model_http.go +++ b/internal/platform/canvas_model_http.go @@ -172,6 +172,11 @@ func (s *service) getCanvasModel(w http.ResponseWriter, r *http.Request, project if err = tx.Rollback(); err != nil { return err } + if r.Header.Get("Accept") == modelStreamType { + writeCanvasModelStream(r.Context(), w, model) + return nil + } + w.Header().Add("Vary", "Accept") reply(w, 200, response{"model": model.view(nil)}) return nil } diff --git a/internal/platform/canvas_model_integration_test.go b/internal/platform/canvas_model_integration_test.go index 0af23ff..9e3995d 100644 --- a/internal/platform/canvas_model_integration_test.go +++ b/internal/platform/canvas_model_integration_test.go @@ -214,6 +214,16 @@ func TestCanonicalModelMigrationAndIncrementalLifecycle(t *testing.T) { if call("GET", base+"/model", nil, "", 200)["model"].(map[string]any)["complete"] != true { t.Fatal("initial read has no explicit coverage") } + // Negotiate the same authenticated canonical read as a streamed transfer. + streamRequest := httptest.NewRequest("GET", base+"/model", nil) + streamRequest.Header.Set("Authorization", "Bearer "+token) + streamRequest.Header.Set("Accept", modelStreamType) + streamWriter := httptest.NewRecorder() + mux.ServeHTTP(streamWriter, streamRequest) + streamed := streamRecords(t, streamWriter) + if streamWriter.Code != 200 || streamWriter.Header().Get("Content-Type") != modelStreamType || len(streamed) < 2 || streamed[0]["projectId"] != root || streamed[len(streamed)-1]["kind"] != "complete" || !modelEqual(streamed[0]["versions"], read().Versions) { + t.Fatal("authenticated stream lost canonical snapshot coverage") + } // An object created and removed inside one transaction still has a lifecycle fact. ghostID := modelTestID(4000) ghost := modelTestNode(ghostID, root) diff --git a/internal/platform/canvas_model_stream.go b/internal/platform/canvas_model_stream.go new file mode 100644 index 0000000..8bc9047 --- /dev/null +++ b/internal/platform/canvas_model_stream.go @@ -0,0 +1,91 @@ +package platform + +import ( + "context" + "encoding/json" + "net/http" +) + +const modelStreamType = "application/vnd.foundry.canvas-model+ndjson;version=1" + +// The caller captures a coherent model while holding the workspace lock, then +// releases its transaction before writing. A slow reader must not hold a writer. +// This streams transfer/encoding; it does not stream database row acquisition. +func writeCanvasModelStream(ctx context.Context, w http.ResponseWriter, model canvasModel) { + token := newModelID() + if token == "" { + reject(w, 500, "model_stream_unavailable") + return + } + view := model.view(nil) + header := response{"streamVersion": 1, "kind": "header", "sequence": 0, "snapshotToken": token, + "documentFormatVersion": canvasModelFormat, "projectId": model.Root, "canvases": view.Canvases, + "versions": view.Versions, "tombstones": view.Tombstones, "removedCanvasIds": view.Removed, "elementCount": len(view.Elements)} + const maxLine = 8 * 1024 * 1024 + const maxBytes = 64 * 1024 * 1024 + sequence, total := 0, 0 + started := false + write := func(value response) bool { + if ctx.Err() != nil { + return false + } + raw, err := json.Marshal(value) + if err != nil || len(raw) > maxLine || total+len(raw)+1 > maxBytes || sequence >= 4096 { + if !started { + reject(w, 413, "model_stream_budget") + } + return false + } + if !started { + w.Header().Set("Content-Type", modelStreamType) + w.Header().Set("Cache-Control", "no-store") + w.Header().Add("Vary", "Accept") + w.Header().Set("X-Content-Type-Options", "nosniff") + w.WriteHeader(http.StatusOK) + started = true + } + total += len(raw) + 1 + if n, writeErr := w.Write(append(raw, '\n')); writeErr != nil || n != len(raw)+1 { + return false + } + sequence++ + // A missing terminal record is an explicit incomplete stream at the client. + // Never append an unrelated JSON error after streaming has begun. + return http.NewResponseController(w).Flush() == nil + } + if !write(header) { + return + } + batch := []map[string]any{} + batchBytes := 0 + flush := func() bool { + if len(batch) == 0 { + return true + } + if !write(response{"streamVersion": 1, "kind": "elements", "sequence": sequence, "snapshotToken": token, "elements": batch}) { + return false + } + batch, batchBytes = []map[string]any{}, 0 + return true + } + for _, element := range view.Elements { + if ctx.Err() != nil { + return + } + raw, err := json.Marshal(element) + if err != nil { + return + } + if len(batch) > 0 && (len(batch) == 64 || batchBytes+len(raw) > 512*1024) { + if !flush() { + return + } + } + batch = append(batch, element) + batchBytes += len(raw) + } + if !flush() { + return + } + write(response{"streamVersion": 1, "kind": "complete", "sequence": sequence, "snapshotToken": token, "elementCount": len(view.Elements)}) +} diff --git a/internal/platform/canvas_model_stream_test.go b/internal/platform/canvas_model_stream_test.go new file mode 100644 index 0000000..9ddce6d --- /dev/null +++ b/internal/platform/canvas_model_stream_test.go @@ -0,0 +1,92 @@ +package platform + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "net/http/httptest" + "testing" +) + +func streamRecords(t *testing.T, writer *httptest.ResponseRecorder) []response { + t.Helper() + var records []response + scanner := bufio.NewScanner(writer.Body) + for scanner.Scan() { + var record response + if err := json.Unmarshal(scanner.Bytes(), &record); err != nil { + t.Fatal(err) + } + records = append(records, record) + } + if err := scanner.Err(); err != nil { + t.Fatal(err) + } + return records +} +func TestCanonicalModelStreamFramesAndSnapshot(t *testing.T) { + model := modelTestDocument() + for i := 0; i < 129; i++ { + id := fmt.Sprintf("extra-%03d", i) + model.Elements[id] = modelTestNode(id, modelRootID) + } + model.Tombstones["deleted"] = 2 + model.TombstoneOwners["deleted"] = modelRootID + writer := httptest.NewRecorder() + writeCanvasModelStream(context.Background(), writer, model) + if writer.Code != 200 || writer.Header().Get("Content-Type") != modelStreamType || !writer.Flushed { + t.Fatal("stream was not negotiated/flushed", writer.Code, writer.Header()) + } + records := streamRecords(t, writer) + if len(records) != 5 || records[0]["kind"] != "header" || records[4]["kind"] != "complete" { + t.Fatal("missing stream framing", len(records)) + } + token := modelString(records[0]["snapshotToken"]) + if !uuidSlugPattern.MatchString(token) || records[0]["projectId"] != modelRootID || !modelEqual(records[0]["versions"], model.Versions) || !modelEqual(records[0]["tombstones"], model.Tombstones) { + t.Fatal("stream header lost snapshot identity") + } + seen := map[string]bool{} + for i, record := range records { + if record["sequence"] != float64(i) || record["snapshotToken"] != token || record["streamVersion"] != float64(1) { + t.Fatal("mixed or unordered snapshot") + } + if record["kind"] == "elements" { + items := record["elements"].([]any) + if len(items) == 0 || len(items) > 64 { + t.Fatal("unbounded record batch") + } + for _, item := range items { + element := item.(map[string]any) + id := modelString(element["id"]) + if seen[id] || !modelEqual(element, model.Elements[id]) { + t.Fatal("duplicated or corrupted stream element") + } + seen[id] = true + } + } + } + if len(seen) != len(model.Elements) || records[4]["elementCount"] != float64(len(seen)) { + t.Fatal("terminal coverage mismatch") + } +} + +type cancelStreamWriter struct { + *httptest.ResponseRecorder + cancel context.CancelFunc +} + +func (writer cancelStreamWriter) Flush() { + writer.ResponseRecorder.Flush() + writer.cancel() +} +func TestCanonicalModelStreamCancellationNeverClaimsCompletion(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + writer := httptest.NewRecorder() + writeCanvasModelStream(ctx, cancelStreamWriter{writer, cancel}, modelTestDocument()) + records := streamRecords(t, writer) + if len(records) != 1 || records[0]["kind"] != "header" { + t.Fatal("cancelled stream wrote a tail", len(records)) + } +}