From 73f411975660f642f389b278522f5d00b2eda3af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E9=BE=99=E5=AE=89?= <297888591+wanglongan587@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:43:36 +0800 Subject: [PATCH 1/3] test: cover asset import workspace fan-out and user-edit preservation The two import gaps circus#195 names that need no production change: with several live Workspaces the import stays scoped to the space (one source row, one workflow and one published snapshot per document), and a package upgrade after the user edited the imported copy and published their own snapshot changes nothing the user owns. --- .../workflow_asset_import_scope_test.go | 190 ++++++++++++++++++ 1 file changed, 190 insertions(+) create mode 100644 integration/workflow_asset_import_scope_test.go diff --git a/integration/workflow_asset_import_scope_test.go b/integration/workflow_asset_import_scope_test.go new file mode 100644 index 000000000..fea73cac6 --- /dev/null +++ b/integration/workflow_asset_import_scope_test.go @@ -0,0 +1,190 @@ +package integration + +import ( + "testing" + + "github.com/wanglongan587/cloud/internal/core" + "github.com/wanglongan587/cloud/internal/workflowasset" +) + +// TestWorkflowAssetImportDoesNotFanOutToWorkspaces covers the fan-out boundary of +// specs test-cases/cloud/plugin-marketplace/workflow-plugin-assets.md: the import is +// triggered by the space's desired state and stays scoped to it. The zero-Workspace case +// is already covered by TestWorkflowAssetImportFromInstalledPlugin; this is the missing +// complement — with several live Workspaces in the space, the Node-side install still +// fans out (one workspace_plugin_instances row per live runtime), but the import must not +// follow it: one source row, one workflow and one published snapshot per document. +func TestWorkflowAssetImportDoesNotFanOutToWorkspaces(t *testing.T) { + f := setup(t) + repo := workflowMarketplace(t, f.root) + pkg := workflowPackage(t, map[string]string{ + "release.json": executableDocument("Release Kit", ""), + "deploy.json": executableDocument("Deploy Kit", ""), + }) + packages := newWorkflowPackages(t, map[string][]byte{"/packages/fanout-1.0.0.orax": pkg}) + publishWorkflowRelease(t, repo, "fanout-kit", "1.0.0", packages.server.URL+"/packages/fanout-1.0.0.orax", pkg) + f.syncMarketplace(t, repo) + f.store.WorkflowAssets = workflowasset.NewFetcher(nil, packages.server.Client().Transport) + + // Several live Workspaces in the one space: the project's main Workspace plus two + // isolated ones, all ready before the install. + created := f.create("fanout-project") + f.drain() + pid := created.O("resource").S("id") + for _, target := range []struct{ title, key string }{{"Fanout One", "fanout-ws-one"}, {"Fanout Two", "fanout-ws-two"}} { + out := f.call("POST", f.path("/projects/"+pid+"/workspaces"), core.Object{"title": target.title, "baseRef": "main"}, target.key, 202) + f.drain() + wid := out.O("resource").S("id") + if state := f.ws(wid).S("observedState"); state != "ready" { + t.Fatalf("workspace %s must be live before the install, got %q", wid, state) + } + } + + sid := f.defaultSpaceID() + f.installWorkflowPlugin(t, sid, "official/fanout-kit", "") + f.drain() + // The Node-side install fans out to every live runtime; the import must not follow it. + liveWorkspaces := f.scalar("SELECT count(*) FROM workspaces WHERE deleted_at IS NULL") + if liveWorkspaces < 2 { + t.Fatalf("the fixture expected at least two live workspaces, got %d", liveWorkspaces) + } + if n := f.scalar("SELECT count(*) FROM workspace_plugin_instances WHERE identifier='fanout-kit'"); n != liveWorkspaces { + t.Fatalf("the install must fan out to all %d live workspaces, got %d instances", liveWorkspaces, n) + } + + must(t, f.importWorkflowAssets(t)) + item := f.assetImport(t, sid, "official/fanout-kit") + if item == nil || item.S("state") != "completed" || item.S("packageVersion") != "1.0.0" { + t.Fatalf("package import = %v", item) + } + if documents := assetDocuments(item); len(documents) != 2 { + t.Fatalf("documents = %v", documents) + } + // One source row for the whole space, not one per workspace. + if n := f.scalar("SELECT count(*) FROM workflow_asset_imports WHERE identifier='fanout-kit'"); n != 1 { + t.Fatalf("the import fanned out to workspaces: %d source rows", n) + } + // One workflow and one published snapshot per document, despite several live runtimes. + if n := f.scalar("SELECT count(*) FROM workflow_asset_documents WHERE identifier='fanout-kit'"); n != 2 { + t.Fatalf("per-workspace document results appeared: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflows WHERE deleted_at IS NULL"); n != 2 { + t.Fatalf("workflows were created per workspace: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots"); n != 2 { + t.Fatalf("snapshots were published per workspace: %d rows", n) + } + if live := f.liveWorkflows(t); len(live) != 2 || live["Release Kit"] == nil || live["Deploy Kit"] == nil { + t.Fatalf("imported workflows = %v", live) + } + + // A restart-equivalent second pass still creates nothing new. + must(t, f.importWorkflowAssets(t)) + if n := f.scalar("SELECT count(*) FROM workflow_asset_documents"); n != 2 { + t.Fatalf("second pass changed document results: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflows"); n != 2 { + t.Fatalf("second pass created workflows: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots"); n != 2 { + t.Fatalf("second pass published snapshots: %d rows", n) + } +} + +// TestWorkflowAssetImportKeepsUserEditsAcrossUpgrade covers the copy-ownership boundary of +// specs test-cases/cloud/plugin-marketplace/workflow-plugin-assets.md: once a document has +// imported, the copy is user data. After the user edits the imported workflow and publishes +// their own snapshot, a package upgrade imports only paths that never succeeded; the user's +// live graph and every published snapshot keep their exact stored text, and the upgraded +// document's new content appears nowhere. +func TestWorkflowAssetImportKeepsUserEditsAcrossUpgrade(t *testing.T) { + f := setup(t) + repo := workflowMarketplace(t, f.root) + v1 := workflowPackage(t, map[string]string{ + "release.json": executableDocument("Edit Kit", `"origin":"package-1.0.0"`), + "deploy.json": executableDocument("Deploy Kit", ""), + }) + v2 := workflowPackage(t, map[string]string{ + "release.json": executableDocument("Edit Kit", `"origin":"package-1.1.0"`), + "deploy.json": executableDocument("Deploy Kit", ""), + "hotfix.json": executableDocument("Hotfix Kit", ""), + }) + packages := newWorkflowPackages(t, map[string][]byte{ + "/packages/edit-kit-1.0.0.orax": v1, + "/packages/edit-kit-1.1.0.orax": v2, + }) + publishWorkflowRelease(t, repo, "edit-kit", "1.0.0", packages.server.URL+"/packages/edit-kit-1.0.0.orax", v1) + f.syncMarketplace(t, repo) + f.store.WorkflowAssets = workflowasset.NewFetcher(nil, packages.server.Client().Transport) + + sid := f.defaultSpaceID() + f.installWorkflowPlugin(t, sid, "official/edit-kit", "") + must(t, f.importWorkflowAssets(t)) + documents := assetDocuments(f.assetImport(t, sid, "official/edit-kit")) + editID := documents["assets/workflows/release.json"].S("workflowId") + if editID == "" || documents["assets/workflows/deploy.json"].S("workflowId") == "" { + t.Fatalf("documents were not imported: %v", documents) + } + + // The user edits the imported copy and publishes their own second snapshot. + edited := f.liveWorkflows(t)["Edit Kit"] + editedGraph := core.Object{"nodes": []any{map[string]any{"id": "user-start", "data": map[string]any{"kind": "start"}}}, "edges": []any{}} + f.call("PUT", f.path("/workflows/"+editID), core.Object{"graph": editedGraph, "version": edited.N("version")}, "edit-user-edit", 200) + f.call("POST", f.path("/workflows/"+editID+"/publish"), core.Object{}, "edit-user-publish", 200) + snapshotText := func(version int64) string { + t.Helper() + var text string + must(t, f.store.Pool.QueryRow("SELECT graph::text FROM workflow_snapshots WHERE workflow_id=$1 AND version=$2", editID, version).Scan(&text)) + return text + } + liveText := func() string { + t.Helper() + var text string + must(t, f.store.Pool.QueryRow("SELECT graph::text FROM workflows WHERE id=$1", editID).Scan(&text)) + return text + } + liveBefore, importSnapshotBefore, userSnapshotBefore := liveText(), snapshotText(1), snapshotText(2) + + // Upgrade the package: the already-imported path now carries different content. + publishWorkflowRelease(t, repo, "edit-kit", "1.1.0", packages.server.URL+"/packages/edit-kit-1.1.0.orax", v2) + f.syncMarketplace(t, repo) + f.installWorkflowPlugin(t, sid, "official/edit-kit", "1.1.0") + must(t, f.importWorkflowAssets(t)) + + documents = assetDocuments(f.assetImport(t, sid, "official/edit-kit")) + if len(documents) != 3 || documents["assets/workflows/hotfix.json"].S("outcome") != "imported" { + t.Fatalf("the upgrade must import only the new path: %v", documents) + } + // The imported path keeps pointing at the copy the user now owns. + if documents["assets/workflows/release.json"].S("workflowId") != editID { + t.Fatalf("the upgrade revisited an imported path: %v", documents["assets/workflows/release.json"]) + } + if live := f.liveWorkflows(t); len(live) != 3 || live["Hotfix Kit"] == nil { + t.Fatalf("workflow library after upgrade = %v", live) + } + // The user's live graph and both published snapshots keep their exact stored text. + if after := liveText(); after != liveBefore { + t.Fatalf("the upgrade overwrote the user's edit: %s -> %s", liveBefore, after) + } + if after := snapshotText(1); after != importSnapshotBefore { + t.Fatalf("the upgrade rewrote the imported snapshot: %s -> %s", importSnapshotBefore, after) + } + if after := snapshotText(2); after != userSnapshotBefore { + t.Fatalf("the upgrade rewrote the user's snapshot: %s -> %s", userSnapshotBefore, after) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots WHERE workflow_id=$1", editID); n != 2 { + t.Fatalf("the upgrade must not add snapshots: %d rows", n) + } + // The upgraded document's new content appears nowhere. + if n := f.scalar("SELECT count(*) FROM workflows WHERE graph::text LIKE '%package-1.1.0%'"); n != 0 { + t.Fatalf("package upgrade content leaked into a live graph: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots WHERE graph::text LIKE '%package-1.1.0%'"); n != 0 { + t.Fatalf("package upgrade content leaked into a snapshot: %d rows", n) + } + // The untouched sibling keeps exactly its own one snapshot. + deployID := documents["assets/workflows/deploy.json"].S("workflowId") + if n := f.scalar("SELECT count(*) FROM workflow_snapshots WHERE workflow_id=$1", deployID); n != 1 { + t.Fatalf("the sibling's snapshots changed: %d rows", n) + } +} From 0456622a91cde9e8a27d85098c7e7eb2acb331fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E9=BE=99=E5=AE=89?= <297888591+wanglongan587@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:43:36 +0800 Subject: [PATCH 2/3] feat(modelgateway): add a trust-root constructor for hermetic transport tests NewUpstreamClientWithRoots builds the production upstream client with an explicit TLS root pool instead of the platform pool, following the existing gitrefs.NewWithHTTP seam. Production wiring keeps calling NewUpstreamClient, so deployed upstream identity still comes from the deployment's system roots; tests that must run the checked dialer, the 10/30-second deadlines and TLS hostname verification against a controlled HTTPS server inject the root that signed it. The roots travel as a constructor argument rather than an UpstreamConfig field so the deployment configuration struct keeps its existing shape and size; modelgateway always receives the full UpstreamConfig by value, and growing it past the gocritic hugeParam threshold would force a pointer signature change on the production path. --- internal/modelgateway/transport.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/internal/modelgateway/transport.go b/internal/modelgateway/transport.go index d0db9bf3d..e3c6c8a19 100644 --- a/internal/modelgateway/transport.go +++ b/internal/modelgateway/transport.go @@ -2,6 +2,8 @@ package modelgateway import ( "context" + "crypto/tls" + "crypto/x509" "errors" "net" "net/http" @@ -83,6 +85,21 @@ type UpstreamConfig struct { // NewUpstreamClient disables environmental proxies and redirects, and enforces the deployment // policy on every request and new connection. Exact fixture host exemptions are deployment-owned, development only. func NewUpstreamClient(config UpstreamConfig) (*http.Client, error) { + return newUpstreamClient(config, nil) +} + +// NewUpstreamClientWithRoots builds the same client but verifies upstream identity against roots +// instead of the platform pool. Hermetic transport tests use it against a controlled HTTPS server +// and keep every other check exactly as deployed; production wiring only calls NewUpstreamClient, +// so upstream identity always comes from the deployment's system roots. +func NewUpstreamClientWithRoots(config UpstreamConfig, roots *x509.CertPool) (*http.Client, error) { + if roots == nil { + return nil, errors.New("upstream TLS trust requires an explicit root pool") + } + return newUpstreamClient(config, roots) +} + +func newUpstreamClient(config UpstreamConfig, roots *x509.CertPool) (*http.Client, error) { if config.Policy.Enterprise() && len(config.DevelopmentHosts) != 0 { return nil, errors.New("enterprise egress conflicts with development hosts") } @@ -103,6 +120,9 @@ func NewUpstreamClient(config UpstreamConfig) (*http.Client, error) { } d := &secureDialer{resolver: resolver, fixtures: net.DefaultResolver, dialer: net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}, allowed: allowed, policy: config.Policy} transport := &http.Transport{Proxy: nil, DialContext: d.dial, TLSHandshakeTimeout: 10 * time.Second, ResponseHeaderTimeout: 30 * time.Second, IdleConnTimeout: 60 * time.Second, MaxIdleConns: 64, MaxConnsPerHost: 32, MaxResponseHeaderBytes: 64 << 10, ForceAttemptHTTP2: !config.Policy.Enterprise()} + if roots != nil { + transport.TLSClientConfig = &tls.Config{RootCAs: roots} + } return &http.Client{Transport: &upstreamTransport{Transport: transport, resolver: resolver, policy: config.Policy}, CheckRedirect: func(*http.Request, []*http.Request) error { return endpointRedirectForbidden }}, nil } From 23e4078da67ac9ca570b3bb951f6bc4276e0859c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E9=BE=99=E5=AE=89?= <297888591+wanglongan587@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:43:36 +0800 Subject: [PATCH 3/3] test: trigger the asset fetch time and size bounds end to end The four approved fetch bounds had no slow-response evidence: the 10s connection deadline, the 30s response-header deadline, the 60s whole-package deadline and the 16 MiB streaming cap are now each triggered for real through the production upstream transport against a controlled HTTPS server (public deployment egress, fixture-host exemption for 127.0.0.1 only, TLS verified against a generated CA). Every case asserts the package-level failure leaves no document, workflow or snapshot rows and that the next administrator trigger retries the same URL successfully. gin.SetMode moves behind a sync.Once so parallel subtests can each build their own fixture; the whole test costs its longest boundary (60s), not the sum. --- integration/cloud_test.go | 7 +- .../workflow_asset_fetch_boundaries_test.go | 347 ++++++++++++++++++ 2 files changed, 353 insertions(+), 1 deletion(-) create mode 100644 integration/workflow_asset_fetch_boundaries_test.go diff --git a/integration/cloud_test.go b/integration/cloud_test.go index 96728c777..835774a31 100644 --- a/integration/cloud_test.go +++ b/integration/cloud_test.go @@ -96,6 +96,11 @@ func testSchema(t *testing.T, prefix string) (*sql.DB, *pgx.ConnConfig) { return pool, config } +// setGinTestModeOnce makes setup safe to call from parallel tests: gin.SetMode writes a +// package global, so concurrent setups would race under the race detector. The first call +// wins and every router is created after the mode is fixed. +var setGinTestModeOnce sync.Once + func setup(t *testing.T) *fixture { t.Helper() pool, config := testSchema(t, "test_") @@ -116,7 +121,7 @@ func setup(t *testing.T) *fixture { must(t, e) auth, e := core.NewAuthenticator("ora-cloud", credentials.Trust) must(t, e) - gin.SetMode(gin.TestMode) + setGinTestModeOnce.Do(func() { gin.SetMode(gin.TestMode) }) log, _ := zap.NewDevelopment() cloud := httptest.NewServer(router.NewDevelopment(store, auth, log)) t.Cleanup(cloud.Close) diff --git a/integration/workflow_asset_fetch_boundaries_test.go b/integration/workflow_asset_fetch_boundaries_test.go new file mode 100644 index 000000000..87f0e9b6a --- /dev/null +++ b/integration/workflow_asset_fetch_boundaries_test.go @@ -0,0 +1,347 @@ +package integration + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "io" + "math/big" + "net" + "net/http" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/wanglongan587/cloud/internal/egress" + "github.com/wanglongan587/cloud/internal/modelgateway" + "github.com/wanglongan587/cloud/internal/workflowasset" +) + +// boundaryMode selects one served package behavior. The modes map one-to-one onto the +// approved fetch bounds: connection establishment, response headers, the whole-package +// deadline, the streaming size bound, and the repaired package for the retry. +type boundaryMode string + +const ( + boundaryHoldTLS boundaryMode = "hold-tls" + boundaryNoHeaders boundaryMode = "no-headers" + boundaryDribble boundaryMode = "dribble" + boundaryOversize boundaryMode = "oversize" + boundaryGood boundaryMode = "good" +) + +// boundaryOversizePlanned is far past the 16 MiB streaming bound plus every socket buffer +// involved, so the server can never finish writing it after the client stops reading. +const boundaryOversizePlanned int64 = 128 << 20 + +// testUpstreamIdentity is the CA and leaf the boundary server presents. The fetcher's +// transport is constructed with the CA in its root pool, so the boundary tests verify +// upstream TLS identity exactly as production does, without touching platform roots. +type testUpstreamIdentity struct { + pool *x509.CertPool + leaf tls.Certificate +} + +func newUpstreamIdentity(t *testing.T) testUpstreamIdentity { + t.Helper() + caKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + must(t, err) + caTemplate := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "ora-asset-boundary-ca"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + IsCA: true, + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + BasicConstraintsValid: true, + } + caDER, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, &caKey.PublicKey, caKey) + must(t, err) + ca, err := x509.ParseCertificate(caDER) + must(t, err) + pool := x509.NewCertPool() + pool.AddCert(ca) + + leafKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + must(t, err) + leafTemplate := &x509.Certificate{ + SerialNumber: big.NewInt(2), + Subject: pkix.Name{CommonName: "127.0.0.1"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + leafDER, err := x509.CreateCertificate(rand.Reader, leafTemplate, ca, &leafKey.PublicKey, caKey) + must(t, err) + return testUpstreamIdentity{pool: pool, leaf: tls.Certificate{Certificate: [][]byte{leafDER}, PrivateKey: leafKey}} +} + +// boundaryServer is the controlled HTTPS server the fetch-boundary tests point the catalog +// at. It owns one TCP port for its whole life so a catalog URL stays valid across the +// failure and the repaired retry. Every connection goes through the production transport's +// real checks: DNS is exempted only for the fixture host 127.0.0.1, TLS identity is +// verified against the generated CA, and the response behavior is exactly the selected +// mode. The mode switch is what models a publisher repairing a broken release. +type boundaryServer struct { + tlsCfg *tls.Config + ln net.Listener + mode atomic.Value + good []byte + + // written and interrupted report what the oversize stream actually reached: the byte + // count the socket accepted, and a one-time signal that a write failed before the + // planned end. They are the server-side half of the "interrupted before the end" + // evidence; the client-side half is the bounded failure code. + written atomic.Int64 + interrupted chan struct{} + failOnce sync.Once +} + +func newBoundaryServer(t *testing.T, identity testUpstreamIdentity, good []byte, mode boundaryMode) *boundaryServer { + t.Helper() + s := &boundaryServer{ + tlsCfg: &tls.Config{Certificates: []tls.Certificate{identity.leaf}}, + good: good, + interrupted: make(chan struct{}), + } + s.mode.Store(mode) + ln, err := net.Listen("tcp", "127.0.0.1:0") + must(t, err) + s.ln = ln + mux := http.NewServeMux() + mux.HandleFunc("/package.orax", s.serve) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go s.handle(conn, mux) + } + }() + t.Cleanup(func() { _ = ln.Close() }) + return s +} + +func (s *boundaryServer) url() string { return "https://" + s.ln.Addr().String() + "/package.orax" } +func (s *boundaryServer) switchTo(m boundaryMode) { s.mode.Store(m) } + +// handle routes one accepted connection. The hold-tls mode holds the raw connection +// without any TLS bytes, so the transport's 10-second connection deadline must cancel the +// fetch; every other mode completes the TLS handshake and serves real HTTP over it. +func (s *boundaryServer) handle(conn net.Conn, mux *http.ServeMux) { + if s.mode.Load() == boundaryHoldTLS { + // The client speaks first in TLS; drain its ClientHello and stay silent. The + // deadline bounds the hold so a lost client cannot leak the goroutine. + _ = conn.SetDeadline(time.Now().Add(2 * time.Minute)) + _, _ = io.Copy(io.Discard, conn) + _ = conn.Close() + return + } + server := &http.Server{Handler: mux, ReadHeaderTimeout: 10 * time.Second} + _ = server.Serve(&singleConnListener{conn: tls.Server(conn, s.tlsCfg)}) +} + +// singleConnListener hands exactly one connection to http.Server.Serve. Each accepted +// connection gets its own server instance, so nothing is shared between connections. +type singleConnListener struct { + conn net.Conn + once sync.Once +} + +func (l *singleConnListener) Accept() (net.Conn, error) { + var conn net.Conn + l.once.Do(func() { conn = l.conn }) + if conn == nil { + return nil, net.ErrClosed + } + return conn, nil +} + +func (l *singleConnListener) Close() error { return nil } +func (l *singleConnListener) Addr() net.Addr { return l.conn.LocalAddr() } + +func (s *boundaryServer) serve(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/octet-stream") + switch s.mode.Load() { + case boundaryNoHeaders: + // ResponseHeaderTimeout must be what fires, so nothing is written: no status, no + // headers. The wait ends when the cancelled client closes the connection. + select { + case <-r.Context().Done(): + case <-time.After(2 * time.Minute): + } + case boundaryDribble: + // Headers arrive immediately; the body trickles far slower than the 60-second + // whole-package deadline can consume, and never finishes inside it. + w.WriteHeader(http.StatusOK) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-r.Context().Done(): + return + case <-ticker.C: + if _, err := w.Write([]byte("x")); err != nil { + return + } + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + } + } + case boundaryOversize: + // No Content-Length: the declared-size refusal cannot fire, so only the streaming + // count can interrupt the download. The planned volume is far beyond every buffer + // in the path, so a client that stops reading at the bound leaves the server + // blocked and then failed before it can finish writing. + w.WriteHeader(http.StatusOK) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + chunk := make([]byte, 64<<10) + for s.written.Load() < boundaryOversizePlanned { + n, err := w.Write(chunk) + s.written.Add(int64(n)) + if err != nil { + s.failOnce.Do(func() { close(s.interrupted) }) + return + } + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + } + default: + w.WriteHeader(http.StatusOK) + _, _ = w.Write(s.good) + } +} + +// realAssetFetcher wires the fetcher over the production upstream transport, not the test +// server's own client: the 10-second connection deadline, the 30-second response-header +// deadline and the checked dialer all come from modelgateway exactly as cmd/server wires +// them. The fixture-host exemption admits only 127.0.0.1, the public deployment policy keeps +// every production gate on, and the injected root pool keeps TLS verification real. +func realAssetFetcher(t *testing.T, identity testUpstreamIdentity) *workflowasset.Fetcher { + t.Helper() + policy, err := egress.Parse([]byte(`{"schemaVersion":1,"mode":"public"}`)) + must(t, err) + client, err := modelgateway.NewUpstreamClientWithRoots(modelgateway.UpstreamConfig{ + DevelopmentHosts: []string{"127.0.0.1"}, + DNS: egress.DefaultDNSConfig(), + Policy: policy, + }, identity.pool) + must(t, err) + t.Cleanup(client.CloseIdleConnections) + return workflowasset.NewFetcher(nil, client.Transport) +} + +// TestWorkflowAssetImportCancelsSlowOrOversizePackages covers the fetch-boundary row of +// specs test-cases/cloud/plugin-marketplace/workflow-plugin-assets.md: slow or oversized +// packages are cancelled by the real transport at the approved 10/30/60-second deadlines +// and the 16 MiB streaming bound, the failure is package-level (no document rows, no +// workflows), and the next administrator trigger retries the same URL and succeeds once +// the publisher's storage is repaired. Each case runs through the production transport +// chain; the subtests share nothing (own schema, marketplace and server) and run in +// parallel, so the test costs its longest boundary rather than their sum. +func TestWorkflowAssetImportCancelsSlowOrOversizePackages(t *testing.T) { + cases := []struct { + name string + mode boundaryMode + code string + // min and max bracket the deadline that must have cancelled the fetch: the lower + // bound proves no earlier failure fired, the upper bound (with generous CI slack) + // distinguishes this deadline from the next-larger one. + min, max time.Duration + }{ + {"connection never established within ten seconds", boundaryHoldTLS, workflowasset.FailureUnavailable, 10 * time.Second, 30 * time.Second}, + {"response headers never arrive within thirty seconds", boundaryNoHeaders, workflowasset.FailureUnavailable, 30 * time.Second, 50 * time.Second}, + {"whole package never completes within sixty seconds", boundaryDribble, workflowasset.FailureUnavailable, 60 * time.Second, 80 * time.Second}, + {"streamed package interrupted past the sixteen MiB bound", boundaryOversize, workflowasset.FailureTooLarge, 0, 0}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + t.Parallel() + f := setup(t) + // Two documents, so a package-level failure must leave neither: even the + // document that would import fine proves the failure is not per-document. + good := workflowPackage(t, map[string]string{ + "release.json": executableDocument("Release Kit", ""), + "deploy.json": executableDocument("Deploy Kit", ""), + }) + repo := workflowMarketplace(t, f.root) + identity := newUpstreamIdentity(t) + server := newBoundaryServer(t, identity, good, c.mode) + publishWorkflowRelease(t, repo, "slow-kit", "1.0.0", server.url(), good) + f.syncMarketplace(t, repo) + f.store.WorkflowAssets = realAssetFetcher(t, identity) + + sid := f.defaultSpaceID() + f.installWorkflowPlugin(t, sid, "official/slow-kit", "") + start := time.Now() + err := f.importWorkflowAssets(t) + if err == nil || !strings.Contains(err.Error(), c.code) { + t.Fatalf("slow package must surface its bounded code %s, got %v", c.code, err) + } + elapsed := time.Since(start) + if elapsed < c.min { + t.Fatalf("the fetch was cancelled early at %s, before its %s deadline", elapsed, c.min) + } + if c.max > 0 && elapsed > c.max { + t.Fatalf("the fetch took %s, past the slack for its %s deadline", elapsed, c.min) + } + item := f.assetImport(t, sid, "official/slow-kit") + if item == nil || item.S("state") != "failed" || item.S("failureCode") != c.code { + t.Fatalf("package failure = %v", item) + } + if n := f.scalar("SELECT count(*) FROM workflow_asset_documents"); n != 0 { + t.Fatalf("a package failure must not record document outcomes: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflows"); n != 0 { + t.Fatalf("a package failure must not create workflows: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots"); n != 0 { + t.Fatalf("a package failure must not publish snapshots: %d rows", n) + } + if c.mode == boundaryOversize { + // The client stopped reading past the bound; the server must observe the + // interrupted stream once the closed socket refuses further writes. + select { + case <-server.interrupted: + case <-time.After(10 * time.Second): + t.Fatal("the server finished writing; the client did not interrupt the stream") + } + if written := server.written.Load(); written < (16<<20)+1 || written >= boundaryOversizePlanned { + t.Fatalf("the server wrote %d bytes; the interruption must land past the bound and before the planned end", written) + } + } + + // The publisher repairs the same URL; the next administrator trigger retries + // and succeeds against the identical catalog selection. + server.switchTo(boundaryGood) + f.installWorkflowPlugin(t, sid, "official/slow-kit", "1.0.0") + must(t, f.importWorkflowAssets(t)) + item = f.assetImport(t, sid, "official/slow-kit") + if item == nil || item.S("state") != "completed" || item.S("failureCode") != "" { + t.Fatalf("retried package = %v", item) + } + documents := assetDocuments(item) + if len(documents) != 2 { + t.Fatalf("retried documents = %v", documents) + } + if live := f.liveWorkflows(t); len(live) != 2 || live["Release Kit"] == nil || live["Deploy Kit"] == nil { + t.Fatalf("retried import did not create the workflows: %v", live) + } + }) + } +}