carlos / platform Public

deploy: watch sweeps targets concurrently

Paul Campbell 5e975efb0440a0712b9889710324fc9917ed6f0e parent 9ea4a0c
2 files changed, +115 −48
  • deploy.go +74 −48
  • deploy_test.go +41 −0
diff --git a/deploy.go b/deploy.go
index b4158f024..5eadd7776 100644
--- a/deploy.go
+++ b/deploy.go
@@ -17,6 +17,8 @@ import (
"os/exec"
"sort"
"strings"
+ "sync"
+ "sync/atomic"
"time"
)
@@ -253,66 +255,90 @@ var watchOrigin = func(host string) string { return "https://" + host }
// deploy as failed. The attempt counter is sufficient (no randomness
// needed): each request just needs to differ from the ones before it.
func watchVersion(w io.Writer, targets []watchTarget, ver string, timeout, interval time.Duration) error {
- pending := make(map[string]watchTarget, len(targets))
- for _, t := range targets {
- pending[t.Host] = t
- }
+ // pending is a slice, not a map, so target order stays stable across
+ // sweeps — the goroutines below write into an indexed result slice keyed
+ // off this same order, and the "live" lines print in that order once the
+ // whole sweep has finished, never interleaved mid-flight.
+ pending := make([]watchTarget, len(targets))
+ copy(pending, targets)
last := make(map[string]string, len(targets))
client := &http.Client{Timeout: 10 * time.Second}
deadline := time.Now().Add(timeout)
- attempt := 0
+ var attempt atomic.Int64 // per-sweep-and-target: every goroutine gets a distinct probe value
for {
- for host, t := range pending {
- attempt++
- url := fmt.Sprintf("%s/?carlos-deploy-probe=%d", t.Origin, attempt)
- resp, err := client.Get(url)
- if err != nil {
- last[host] = "unreachable: " + err.Error()
- continue
- }
- io.CopyN(io.Discard, resp.Body, 4096)
- resp.Body.Close()
- got := resp.Header.Get("X-Carlos-Version")
- switch {
- case got == ver && resp.StatusCode/100 == 2:
+ type result struct {
+ live bool
+ diag string
+ }
+ results := make([]result, len(pending))
+ var wg sync.WaitGroup
+ for i, t := range pending {
+ wg.Add(1)
+ go func(i int, t watchTarget) {
+ defer wg.Done()
+ n := attempt.Add(1)
+ url := fmt.Sprintf("%s/?carlos-deploy-probe=%d", t.Origin, n)
+ resp, err := client.Get(url)
+ if err != nil {
+ results[i] = result{diag: "unreachable: " + err.Error()}
+ return
+ }
+ io.CopyN(io.Discard, resp.Body, 4096)
+ resp.Body.Close()
+ got := resp.Header.Get("X-Carlos-Version")
+ switch {
+ case got == ver && resp.StatusCode/100 == 2:
+ results[i] = result{live: true}
+ case got == ver:
+ // The edge stamps X-Carlos-Version before proxying to the
+ // origin, so a tenant that dies on boot can still answer
+ // with the new version header attached to a non-2xx (a 502,
+ // most commonly). That is the new build in a crash loop, not
+ // a live deploy, so the success arm has to gate on status
+ // too — matching the header alone let a dead tenant exit 0.
+ results[i] = result{diag: fmt.Sprintf("serving %s but HTTP %d — the app is failing", ver, resp.StatusCode)}
+ case got == "" && resp.StatusCode/100 != 2:
+ // A non-2xx with no header is not an old edge — it is an edge
+ // that answered and had nothing good to serve: an activator
+ // 503 because the wake failed or timed out, or a 404 because
+ // the route has not been minted yet. (A 502 from a tenant
+ // that died on boot carries the version header — the edge
+ // stamps it before proxying — so that shape lands in the
+ // case above, not here.) Reporting those as "edge predates
+ // this feature — the hourly pass will adopt it" sends the
+ // operator off to wait for a pass that will not help (review
+ // finding 2): the status code is the diagnostic, so print it.
+ results[i] = result{diag: fmt.Sprintf("not serving (HTTP %d) — instance failing to wake, or route not live yet", resp.StatusCode)}
+ case got == "":
+ // 2xx and no header: something IS being served here, it just
+ // isn't stamped — the only shape that really is an edge from
+ // before X-Carlos-Version existed.
+ results[i] = result{diag: "no X-Carlos-Version header (edge predates this feature — the hourly pass will adopt it)"}
+ default:
+ results[i] = result{diag: "still serving " + got}
+ }
+ }(i, t)
+ }
+ wg.Wait()
+
+ var stillPending []watchTarget
+ for i, r := range results {
+ host := pending[i].Host
+ if r.live {
fmt.Fprintf(w, "%s https://%s\n", colorFor(w).ok("live"), host)
- delete(pending, host)
- case got == ver:
- // The edge stamps X-Carlos-Version before proxying to the
- // origin, so a tenant that dies on boot can still answer
- // with the new version header attached to a non-2xx (a 502,
- // most commonly). That is the new build in a crash loop, not
- // a live deploy, so the success arm has to gate on status
- // too — matching the header alone let a dead tenant exit 0.
- last[host] = fmt.Sprintf("serving %s but HTTP %d — the app is failing", ver, resp.StatusCode)
- case got == "" && resp.StatusCode/100 != 2:
- // A non-2xx with no header is not an old edge — it is an edge
- // that answered and had nothing good to serve: an activator
- // 503 because the wake failed or timed out, or a 404 because
- // the route has not been minted yet. (A 502 from a tenant
- // that died on boot carries the version header — the edge
- // stamps it before proxying — so that shape lands in the
- // case above, not here.) Reporting those as "edge predates
- // this feature — the hourly pass will adopt it" sends the
- // operator off to wait for a pass that will not help (review
- // finding 2): the status code is the diagnostic, so print it.
- last[host] = fmt.Sprintf("not serving (HTTP %d) — instance failing to wake, or route not live yet", resp.StatusCode)
- case got == "":
- // 2xx and no header: something IS being served here, it just
- // isn't stamped — the only shape that really is an edge from
- // before X-Carlos-Version existed.
- last[host] = "no X-Carlos-Version header (edge predates this feature — the hourly pass will adopt it)"
- default:
- last[host] = "still serving " + got
+ continue
}
+ last[host] = r.diag
+ stillPending = append(stillPending, pending[i])
}
+ pending = stillPending
if len(pending) == 0 {
return nil
}
if time.Now().After(deadline) {
hosts := make([]string, 0, len(pending))
- for h := range pending {
- hosts = append(hosts, h)
+ for _, t := range pending {
+ hosts = append(hosts, t.Host)
}
sort.Strings(hosts)
var b strings.Builder
diff --git a/deploy_test.go b/deploy_test.go
index 1f7dac812..500d2fc6f 100644
--- a/deploy_test.go
+++ b/deploy_test.go
@@ -171,6 +171,47 @@ func TestWatchVersionNonOKNoHeaderDiagnostic(t *testing.T) {
}
}
+// TestWatchVersionSweepsTargetsConcurrently pins the concurrency fix: with N
+// targets, a single tick must probe them all in parallel, not one after
+// another. Three origins each sleep 500ms before answering (already live, so
+// one sweep is enough) — sequential probing would take >=1.5s for the sweep
+// alone, parallel finishes in ~500ms. The 1.2s budget leaves generous margin
+// for CI noise while still failing hard on a regression to sequential polling.
+func TestWatchVersionSweepsTargetsConcurrently(t *testing.T) {
+ const delay = 500 * time.Millisecond
+ newSrv := func() *httptest.Server {
+ return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ time.Sleep(delay)
+ w.Header().Set("X-Carlos-Version", "new5678")
+ }))
+ }
+ srvs := []*httptest.Server{newSrv(), newSrv(), newSrv()}
+ defer func() {
+ for _, s := range srvs {
+ s.Close()
+ }
+ }()
+ targets := make([]watchTarget, len(srvs))
+ for i, s := range srvs {
+ targets[i] = watchTarget{Host: fmt.Sprintf("h%d.test", i), Origin: s.URL}
+ }
+ var out bytes.Buffer
+ start := time.Now()
+ err := watchVersion(&out, targets, "new5678", 5*time.Second, 10*time.Millisecond)
+ elapsed := time.Since(start)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if elapsed >= 1200*time.Millisecond {
+ t.Fatalf("sweep took %v — targets were probed sequentially, not concurrently", elapsed)
+ }
+ for _, target := range targets {
+ if !strings.Contains(out.String(), target.Host) {
+ t.Fatalf("live line missing for %s: %q", target.Host, out.String())
+ }
+ }
+}
+
// TestWatchVersionCacheBustsEveryPoll pins the fix for the edge's up-to-5min
// cache TTL on X-Carlos-Version (internal/edge/edge.go, versionHeader): a
// watch with no cache-bust could poll the same cached response for the whole