rogee/creator-hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HH-833: fence gateway generations across replicas (#22)

Browse Source
This commit is contained in:
rogee
2026-08-29 23:27:30 +08:00
parent 9fc8e44123
commit 00d044d440
7 changed files with 83 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/control-plane/hub_test.go
+7 -7
View File
@@ -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)
cmd/docker-gateway/main.go
+15 -13
View File
@@ -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
cmd/docker-gateway/main_test.go
+24 -2
View File
@@ -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) {
cmd/docker-gateway/proxy.go
+12 -3
View File
@@ -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
}
cmd/docker-gateway/proxy_test.go
+22
View File
@@ -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")
}
}