diff --git a/cmd/control-plane/hub_test.go b/cmd/control-plane/hub_test.go index 2def095..43133ca 100644 --- a/cmd/control-plane/hub_test.go +++ b/cmd/control-plane/hub_test.go @@ -451,14 +451,14 @@ func TestPauseSerializesResumeAndRetainsClaimGateOnUnknownStop(t *testing.T) { go func() { pauseDone <- do(app, http.MethodPost, "/api/phase-a/accounts/account-a/pause", "") }() select { case <-blocking.entered: - case <-time.After(time.Second): + case <-time.After(5 * time.Second): t.Fatal("pause did not enter runtime stop") } go func() { resumeDone <- do(app, http.MethodPost, "/api/phase-a/accounts/account-a/resume", "") }() select { case response := <-resumeDone: t.Fatalf("resume bypassed the runtime operation lock: %d", response.Code) - case <-time.After(50 * time.Millisecond): + case <-time.After(time.Second): } close(blocking.release) if response := <-pauseDone; response.Code != http.StatusInternalServerError { @@ -1890,7 +1890,7 @@ func TestPostgresRebindRecoversRealConcurrentRaces(t *testing.T) { }() select { case <-deleteDone: - case <-time.After(time.Second): + case <-time.After(5 * time.Second): close(releaseDelete) t.Fatal("rebind did not reach the DELETE barrier") } @@ -3211,7 +3211,7 @@ func TestImageDisableWaitsForUpgradeCommit(t *testing.T) { }() select { case <-gateway.createStarted: - case <-time.After(time.Second): + case <-time.After(5 * time.Second): t.Fatal("upgrade did not reach gateway create") } @@ -3236,7 +3236,7 @@ func TestImageDisableWaitsForUpgradeCommit(t *testing.T) { select { case result := <-disableDone: t.Fatalf("disable completed before upgrade commit: %#v", result) - case <-time.After(50 * time.Millisecond): + case <-time.After(time.Second): } close(releaseCreate) @@ -3297,7 +3297,7 @@ func TestListAndHeartbeatWaitForUpgradeCoordination(t *testing.T) { }() select { case <-gateway.createStarted: - case <-time.After(time.Second): + case <-time.After(5 * time.Second): t.Fatal("upgrade did not reach gateway create") } @@ -3324,7 +3324,7 @@ func TestListAndHeartbeatWaitForUpgradeCoordination(t *testing.T) { t.Fatalf("list reconciled a stale snapshot during upgrade: %#v", result) case err := <-heartbeatDone: t.Fatalf("heartbeat reconciled a stale snapshot during upgrade: %v", err) - case <-time.After(50 * time.Millisecond): + case <-time.After(time.Second): } close(releaseCreate) diff --git a/cmd/docker-gateway/main.go b/cmd/docker-gateway/main.go index 02fa01f..2e32dea 100644 --- a/cmd/docker-gateway/main.go +++ b/cmd/docker-gateway/main.go @@ -589,9 +589,18 @@ func (api gateway) changeState(c fiber.Ctx) error { var input generationRequest switch action { case "start": - if err := api.requireManaged(id); err != nil { + var err error + input, err = decodeGeneration(c) + if err != nil { + return writeError(c, http.StatusBadRequest, err) + } + _, exists, err := api.requireGeneration(id, input) + if err != nil { return writeError(c, statusFor(err), err) } + if !exists { + return writeError(c, http.StatusNotFound, os.ErrNotExist) + } case "stop": var err error input, err = decodeGeneration(c) @@ -615,10 +624,13 @@ func (api gateway) changeState(c fiber.Ctx) error { defer release() var path string if action == "start" { - containerID, _, err := api.managedContainer(id) + containerID, exists, err := api.requireGeneration(id, input) if err != nil { return writeError(c, statusFor(err), err) } + if !exists { + return writeError(c, http.StatusNotFound, os.ErrNotExist) + } path = "/containers/" + url.PathEscape(containerID) + "/start" } else { containerID, exists, err := api.requireGeneration(id, input) @@ -857,11 +869,6 @@ func (api gateway) requireProxyGeneration(alias string, input proxyRestoreReques return runtimeID, labels, nil } -func (api gateway) requireManaged(id string) error { - _, err := api.managedLabels(id) - return err -} - func decodeGeneration(c fiber.Ctx) (generationRequest, error) { var input generationRequest decoder := json.NewDecoder(bytes.NewReader(c.Body())) @@ -883,17 +890,12 @@ func (api gateway) requireGeneration(id string, input generationRequest) (string return "", false, err } version, _ := strconv.ParseInt(labels[bindingVersionLabel], 10, 64) - if input.RuntimeID == "" || runtimeID != input.RuntimeID || version != input.BindingVersion { + if input.RuntimeID == "" || runtimeID != input.RuntimeID || version != input.BindingVersion || labels[networkIDLabel] != input.NetworkID { return "", false, errGenerationConflict } return runtimeID, true, nil } -func (api gateway) managedLabels(id string) (map[string]string, error) { - _, labels, err := api.managedContainer(id) - return labels, err -} - func (api gateway) managedContainer(id string) (string, map[string]string, error) { if !runtimeIDPattern.MatchString(id) { return "", nil, errInvalidRuntimeID diff --git a/cmd/docker-gateway/main_test.go b/cmd/docker-gateway/main_test.go index d6588ef..dc8b703 100644 --- a/cmd/docker-gateway/main_test.go +++ b/cmd/docker-gateway/main_test.go @@ -1042,7 +1042,11 @@ func TestGatewayLifecycleUsesInspectedImmutableContainerID(t *testing.T) { t.Run(test.method+" "+test.path, func(t *testing.T) { docker, server := testDocker(func(response http.ResponseWriter, request *http.Request) { if request.Method == http.MethodGet { - _, _ = response.Write([]byte(`{"Id":"container-id","Config":{"Labels":{"` + managedLabel + `":"true","` + idLabel + `":"account-a","` + bindingVersionLabel + `":"1"}}}`)) + networkLabel := "" + if test.path == "/v1/browsers/account-a/start" { + networkLabel = `,"` + networkIDLabel + `":"network-id"` + } + _, _ = response.Write([]byte(`{"Id":"container-id","Config":{"Labels":{"` + managedLabel + `":"true","` + idLabel + `":"account-a","` + bindingVersionLabel + `":"1"` + networkLabel + `}}}`)) return } if request.URL.Path != test.dockerPath { @@ -1055,7 +1059,9 @@ func TestGatewayLifecycleUsesInspectedImmutableContainerID(t *testing.T) { handler := newGateway(docker, "creatorhub_browser", testToken) response := httptest.NewRecorder() var body io.Reader - if test.path != "/v1/browsers/account-a/start" { + if test.path == "/v1/browsers/account-a/start" { + body = strings.NewReader(`{"binding_version":1,"runtime_id":"container-id","network_id":"network-id"}`) + } else { body = strings.NewReader(testGenerationBody) } adaptor.FiberApp(handler).ServeHTTP(response, authed(test.method, test.path, body)) @@ -1704,6 +1710,21 @@ func TestGatewayTwoReplicaCreateRestoreRemoveProxyContract(t *testing.T) { proxiesG2.proxies["account-a"].runtimeID != c2 { t.Fatalf("cross-process proxy release contract failed: c1=%q c2=%q g1=%v g2=%v c2Attached=%v", c1, c2, g1Attached, g2Attached, c2Attached) } + state.Lock() + n2 := state.networkID + port = state.containerLabels[proxyPortLabel] + state.Unlock() + restoreBody = `{"binding_version":1,"runtime_id":"` + c2 + `","network_id":"` + n2 + `","network_exit_id":"exit-1","network_exit":{"protocol":"socks5","host":"proxy.example","port":1080}}` + if restored := call(g1, http.MethodPost, "/v1/browsers/account-a/proxy", restoreBody); restored.Code != http.StatusNoContent { + t.Fatalf("G1 successor restore failed: %d %s", restored.Code, restored.Body.String()) + } + state.Lock() + _, g1Attached = state.networkMembers["gateway-g1"] + state.Unlock() + if !g1Attached || proxiesG1.proxies["account-a"].runtimeID != c2 || proxiesG2.proxies["account-a"].runtimeID != c2 { + t.Fatalf("stale replica did not replace its proxy generation: c1=%q c2=%q g1=%v g1Proxy=%v g2Proxy=%v", + c1, c2, g1Attached, proxiesG1.proxies["account-a"], proxiesG2.proxies["account-a"]) + } proxiesG1.remove("account-a", 1, c1) proxiesG2.remove("account-a", 1, c2) } @@ -2177,6 +2198,7 @@ func TestGatewayRejectsStaleGenerationBeforeDockerMutation(t *testing.T) { method, path string }{ {http.MethodPost, "/v1/browsers/account-a/stop"}, + {http.MethodPost, "/v1/browsers/account-a/start"}, {http.MethodDelete, "/v1/browsers/account-a"}, } { t.Run(request.method, func(t *testing.T) { diff --git a/cmd/docker-gateway/proxy.go b/cmd/docker-gateway/proxy.go index a58dde5..bc7f649 100644 --- a/cmd/docker-gateway/proxy.go +++ b/cmd/docker-gateway/proxy.go @@ -53,7 +53,11 @@ func (registry *memoryProxyRegistry) configure(alias string, bindingVersion int6 (port == 0 || proxy.listener.Addr().(*net.TCPAddr).Port == port) && proxy.exit == exit && proxy.networkID == networkID { return proxy.url, func() { registry.removeObject(alias, proxy) }, nil } - return "", nil, errGenerationConflict + delete(registry.proxies, alias) + // A gateway replica may still hold the previous runtime generation. Close + // it before rebinding a restored listener, while removeObject's identity + // check keeps old cleanup callbacks from deleting the replacement. + closeMemoryProxy(proxy) } listener, err := net.Listen("tcp4", net.JoinHostPort(bindHost, strconv.Itoa(port))) if err != nil { @@ -80,10 +84,15 @@ func (registry *memoryProxyRegistry) removeObject(alias string, proxy *memoryPro } registry.mu.Unlock() if proxy != nil { - _ = proxy.server.Close() + closeMemoryProxy(proxy) } } +func closeMemoryProxy(proxy *memoryProxy) { + _ = proxy.listener.Close() + _ = proxy.server.Close() +} + func (registry *memoryProxyRegistry) ready(alias string, port int, runtimeID string, networkIDs ...string) bool { registry.mu.Lock() defer registry.mu.Unlock() @@ -120,7 +129,7 @@ func (registry *memoryProxyRegistry) remove(alias string, bindingVersion int64, } registry.mu.Unlock() if proxy != nil { - _ = proxy.server.Close() + closeMemoryProxy(proxy) } return true } diff --git a/cmd/docker-gateway/proxy_test.go b/cmd/docker-gateway/proxy_test.go index 56264a4..163296a 100644 --- a/cmd/docker-gateway/proxy_test.go +++ b/cmd/docker-gateway/proxy_test.go @@ -168,3 +168,25 @@ func TestMemoryProxyRemoveRequiresMatchingGeneration(t *testing.T) { t.Fatal("stale generation removed the current proxy") } } + +func TestMemoryProxyReplacesStaleGenerationWithoutOldCleanup(t *testing.T) { + registry := newMemoryProxyRegistry() + exit := gatewayProxyExit{Protocol: "http", Host: "127.0.0.1", Port: 1} + oldURL, oldCleanup, err := registry.configure("account-a", 1, "127.0.0.1", 0, exit, "network-n1") + if err != nil { + t.Fatal(err) + } + oldPort := registry.proxies["account-a"].listener.Addr().(*net.TCPAddr).Port + newURL, newCleanup, err := registry.configure("account-a", 2, "127.0.0.1", oldPort, exit, "network-n2") + if err != nil { + t.Fatal(err) + } + defer newCleanup() + if newURL != oldURL || registry.proxies["account-a"].bindingVersion != 2 { + t.Fatalf("stale proxy was not replaced: old=%q new=%q proxy=%#v", oldURL, newURL, registry.proxies["account-a"]) + } + oldCleanup() + if registry.proxies["account-a"].bindingVersion != 2 { + t.Fatal("old cleanup removed the replacement proxy") + } +} diff --git a/docs/architecture/container-control.md b/docs/architecture/container-control.md index aff8060..5039e65 100644 --- a/docs/architecture/container-control.md +++ b/docs/architecture/container-control.md @@ -32,6 +32,7 @@ React ──> control-plane ── /api/browsers ──(Bearer token)──> doc - 只有 `docker-gateway` 挂载 socket，控制面和浏览器容器均不可见；网关加入 control 与 browser 网络，浏览器只拿到无凭据的内存转发代理地址，`/v1` 仍必须通过容器内不可见的网关令牌； - 网关只暴露面向领域的路由，不提供通用 Docker 代理；`/v1` 全部接口校验 `Authorization: Bearer `（常数时间比较），令牌由部署者在网关环境变量与平台注册表中保持一致； +- 网关直连的 `POST /v1/browsers/{alias}/start|stop` 仅供内部维护使用，必须提交并精确匹配容器标签中的 `{binding_version,runtime_id,network_id}`；generation 不匹配返回 `409`，控制面生命周期编排不依赖无 fence 的直连 start； - 网关固定命令、网络、挂载和资源限制；外部输入是受校验的别名，以及平台下发的镜像引用、启动参数和卷名——镜像引用来自平台维护的版本表，新增/变更由人工在页面审核启用，不再写死在代码中； - 启停和删除前必须同时匹配固定名称前缀及 `io.creatorhub.managed`、`io.creatorhub.runtime-id` 标签； - 动态容器使用只读根文件系统、非 root `1000:1000` 与固定镜像入口、全部 capability drop、`no-new-privileges`、CPU/内存/PID 限制，且无宿主机端口和目录挂载； diff --git a/internal/hub/store_test.go b/internal/hub/store_test.go index 55cf2d9..08c6cde 100644 --- a/internal/hub/store_test.go +++ b/internal/hub/store_test.go @@ -338,7 +338,7 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) { if _, err := store.ActivateRuntime(ctx, env.Alias, "container-a", bound.BindingVersion+1, second.ID, "network-a"); !errors.Is(err, ErrConflict) { t.Fatalf("stale binding metadata must not activate a runtime: %v", err) } - if _, err := store.db.ExecContext(ctx, `UPDATE runtime_instance SET lease_until = now() + interval '1 second' WHERE id = $1`, active.RuntimeInstanceID); err != nil { + if _, err := store.db.ExecContext(ctx, `UPDATE runtime_instance SET lease_until = now() + interval '10 seconds' WHERE id = $1`, active.RuntimeInstanceID); err != nil { t.Fatal(err) } if _, err := store.ActivateRuntime(ctx, env.Alias, "container-a", bound.BindingVersion, bound.Exit.ID, "network-a"); err != nil { @@ -440,7 +440,7 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) { select { case err := <-activation: t.Fatalf("activation bypassed the locked account row: %v", err) - case <-time.After(50 * time.Millisecond): + case <-time.After(time.Second): } if _, err := pauseTx.ExecContext(ctx, `UPDATE social_account SET status = 'paused' WHERE id = 'account-a'`); err != nil { t.Fatal(err)