amadan

carlos / platform Public

console: a refused seal, flip or open names the step that stopped; the target-left fact is never dropped; the health probe reuses one transport per target address (and its direct test); witnesses narrowed and the lift's order pinned

Fix round 1, minors 1, 3, 4, 5 and 6.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 844e603bb59219340ad14cae2d7632b66e853051 parent 50eeec8
7 files changed, +193 −16
  • internal/console/selfserve_checks.go +7 −0
  • internal/console/selfserve_finalise_test.go +30 −0
  • internal/console/selfserve_health.go +47 −9
  • internal/console/selfserve_health_test.go +76 −0
  • internal/console/selfserve_job.go +24 −5
  • internal/e2e/selfserve/cut_test.go +5 −0
  • internal/e2e/selfserve/job_boundary_test.go +4 −2
diff --git a/internal/console/selfserve_checks.go b/internal/console/selfserve_checks.go
index 2281f7e31..8093c876e 100644
--- a/internal/console/selfserve_checks.go
+++ b/internal/console/selfserve_checks.go
@@ -40,6 +40,7 @@ const (
factLanding = "the landing"
factOpen = "the open"
factStagePath = "the stage path"
+ factFlip = "the flip"
selfServeReadyMiB = 16 << 20
)
@@ -88,6 +89,12 @@ func ssStagedMismatch(host string) string {
}
func ssReplaced(label string) string { return fmt.Sprintf("box %s was replaced", label) }
+// ssStoppedAt is the abort of a cut write the record refused after the
+// start (step: the seal, the flip or the open; pending copy review).
+func ssStoppedAt(step string) string {
+ return fmt.Sprintf("the move stopped at %s: the instance changed under it", step)
+}
+
// The three target facts Task 8 checks (copy rows pending Paul's review).
const (
msgSelfServeStageUnarmed = "the target box's stage path is not armed"
diff --git a/internal/console/selfserve_finalise_test.go b/internal/console/selfserve_finalise_test.go
index ea1b3f9ca..4f4e2252b 100644
--- a/internal/console/selfserve_finalise_test.go
+++ b/internal/console/selfserve_finalise_test.go
@@ -1,8 +1,11 @@
package console
import (
+ "errors"
"testing"
"time"
+
+ "github.com/carlosframework/platform/internal/provision"
)
// TestSelfServeFinishedRowsNeverAdvance: the finaliser's first step
@@ -77,3 +80,30 @@ func TestSelfServeFinishedRowsNeverAdvance(t *testing.T) {
}
})
}
+
+// TestCutWriteRefusalNamesTheStep: a seal, flip or open the record refused
+// after the start aborts naming the step ("the move stopped at <step>: the
+// instance changed under it"), not "the move could not be started"; any
+// other error is returned and aborts nothing.
+func TestCutWriteRefusalNamesTheStep(t *testing.T) {
+ st := ssStore(t)
+ s := &Server{store: st}
+ now := ssT0.Add(time.Second)
+ for i, step := range []string{factSeal, factFlip, factOpen} {
+ id := []string{"mv_cutrefusedsealcutrefusedse", "mv_cutrefusedflipcutrefusedfl", "mv_cutrefusedopencutrefusedop"}[i]
+ ssAdmit(t, st, ssMove(id, step[4:]+"."+ssHost, ssT0, "ie-1", "uk-1"))
+ row := ssLoad(t, st, id)
+ if err := s.selfServeCutWriteRefused(row, errors.New("a store failure"), step, now); err == nil {
+ t.Fatal("a store failure was swallowed")
+ }
+ if r := ssLoad(t, st, id); !r.AbortRequestedAt.IsZero() {
+ t.Fatal("a store failure aborted the move")
+ }
+ if err := s.selfServeCutWriteRefused(row, &requestRefusal{err: provision.ErrMoveConflict}, step, now); err != nil {
+ t.Fatal(err)
+ }
+ if r := ssLoad(t, st, id); r.OutcomeReason != "the move stopped at "+step+": the instance changed under it" {
+ t.Fatalf("%s: aborted %q", step, r.OutcomeReason)
+ }
+ }
+}
diff --git a/internal/console/selfserve_health.go b/internal/console/selfserve_health.go
index 89124c8d2..d36cda943 100644
--- a/internal/console/selfserve_health.go
+++ b/internal/console/selfserve_health.go
@@ -14,12 +14,15 @@ import (
"context"
"crypto/rand"
"crypto/tls"
+ "crypto/x509"
"encoding/hex"
"fmt"
+ "io"
"math"
"net"
"net/http"
"net/url"
+ "sync"
"time"
"github.com/carlosframework/platform/internal/provision"
@@ -134,7 +137,47 @@ func (s *Server) probeTarget(ctx context.Context, row *SelfServeMove, mv *provis
if _, err := rand.Read(nonce[:]); err != nil {
return false, "no probe value"
}
- u := url.URL{Scheme: "https", Host: row.Host, Path: m.ReadyPathOrDefault(), RawQuery: "carlos-probe=" + hex.EncodeToString(nonce[:])}
+ return probeAt(ctx, eip, row.Host, m.ReadyPathOrDefault(), hex.EncodeToString(nonce[:]), want)
+}
+
+// The probes' transports, one per target address and shared by every
+// probe of it (so a pooled connection is only ever to the box it was
+// dialled to; idle connections are reused and closed by the transport's
+// own timeout). The URL's host is the instance's, so it is the SNI and the
+// Host header.
+var (
+ selfServeProbePort = "443" // a TEST SEAM
+ selfServeProbeRoots *x509.CertPool // a TEST SEAM: nil is the system's
+ probeTransportsMu sync.Mutex
+ probeTransports = map[string]*http.Transport{}
+)
+
+// probeTransport is the transport that dials addr.
+func probeTransport(addr string) *http.Transport {
+ probeTransportsMu.Lock()
+ defer probeTransportsMu.Unlock()
+ if t := probeTransports[addr]; t != nil {
+ return t
+ }
+ dialer := &net.Dialer{Timeout: selfServeProbeTimeout}
+ t := &http.Transport{
+ TLSClientConfig: &tls.Config{RootCAs: selfServeProbeRoots},
+ DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) {
+ return dialer.DialContext(ctx, network, addr)
+ },
+ MaxIdleConnsPerHost: 2,
+ IdleConnTimeout: 90 * time.Second,
+ TLSHandshakeTimeout: selfServeProbeTimeout,
+ }
+ probeTransports[addr] = t
+ return t
+}
+
+// probeAt is one probe of host's ready path on the box at eip: no
+// redirects followed, no cache, 30 seconds; it passes on a 2xx answered by
+// build want.
+func probeAt(ctx context.Context, eip, host, readyPath, value, want string) (bool, string) {
+ u := url.URL{Scheme: "https", Host: host, Path: readyPath, RawQuery: "carlos-probe=" + value}
pctx, cancel := context.WithTimeout(ctx, selfServeProbeTimeout)
defer cancel()
req, err := http.NewRequestWithContext(pctx, http.MethodGet, u.String(), nil)
@@ -142,21 +185,16 @@ func (s *Server) probeTarget(ctx context.Context, row *SelfServeMove, mv *provis
return false, "the probe could not be built"
}
req.Header.Set("Cache-Control", "no-cache")
- dialer := &net.Dialer{Timeout: selfServeProbeTimeout}
client := &http.Client{
- Timeout: selfServeProbeTimeout,
- Transport: &http.Transport{
- TLSClientConfig: &tls.Config{ServerName: row.Host},
- DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) {
- return dialer.DialContext(ctx, network, net.JoinHostPort(eip, "443"))
- },
- },
+ Timeout: selfServeProbeTimeout,
+ Transport: probeTransport(net.JoinHostPort(eip, selfServeProbePort)),
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
}
resp, err := client.Do(req)
if err != nil {
return false, "the probe did not get an answer"
}
+ _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10))
resp.Body.Close()
switch got := resp.Header.Get("x-carlos-version"); {
case resp.StatusCode < 200 || resp.StatusCode > 299:
diff --git a/internal/console/selfserve_health_test.go b/internal/console/selfserve_health_test.go
new file mode 100644
index 000000000..4144b340f
--- /dev/null
+++ b/internal/console/selfserve_health_test.go
@@ -0,0 +1,76 @@
+package console
+
+import (
+ "context"
+ "crypto/x509"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "testing"
+)
+
+// TestProbeAtAsksTheTargetBoxDirectly: the health probe dials the target
+// box's address with the instance's host as SNI and Host, asks the ready
+// path with a probe value and Cache-Control: no-cache, follows no
+// redirect, and passes only on a 2xx from the build the seal recorded; the
+// probes of one address share one transport.
+func TestProbeAtAsksTheTargetBoxDirectly(t *testing.T) {
+ var mu sync.Mutex
+ var seen []*http.Request
+ srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ mu.Lock()
+ seen = append(seen, r.Clone(context.Background()))
+ mu.Unlock()
+ switch r.URL.Path {
+ case "/ready":
+ w.Header().Set("x-carlos-version", "v42")
+ case "/old":
+ w.Header().Set("x-carlos-version", "v41")
+ case "/redirect":
+ w.Header().Set("x-carlos-version", "v42")
+ http.Redirect(w, r, "/ready", http.StatusFound)
+ return
+ case "/down":
+ w.Header().Set("x-carlos-version", "v42")
+ w.WriteHeader(http.StatusServiceUnavailable)
+ return
+ }
+ w.WriteHeader(http.StatusOK)
+ }))
+ defer srv.Close()
+ host, port, _ := net.SplitHostPort(srv.Listener.Addr().String())
+ roots := x509.NewCertPool()
+ roots.AddCert(srv.Certificate())
+ oldPort, oldRoots := selfServeProbePort, selfServeProbeRoots
+ selfServeProbePort, selfServeProbeRoots = port, roots
+ defer func() { selfServeProbePort, selfServeProbeRoots = oldPort, oldRoots }()
+
+ ctx := context.Background()
+ // httptest's certificate names example.com: the instance's host here.
+ for _, c := range []struct {
+ path string
+ ok bool
+ }{{"/ready", true}, {"/old", false}, {"/redirect", false}, {"/down", false}} {
+ if ok, detail := probeAt(ctx, host, "example.com", c.path, "v1", "v42"); ok != c.ok {
+ t.Errorf("%s: %v %q, want %v", c.path, ok, detail, c.ok)
+ }
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ if len(seen) != 4 {
+ t.Fatalf("%d requests (a redirect followed?)", len(seen))
+ }
+ for _, r := range seen {
+ if r.Host != "example.com" || r.TLS == nil || r.TLS.ServerName != "example.com" || r.URL.Query().Get("carlos-probe") != "v1" || r.Header.Get("Cache-Control") != "no-cache" {
+ t.Errorf("request %s host %q sni %v query %q cache %q", r.URL.Path, r.Host, r.TLS, r.URL.RawQuery, r.Header.Get("Cache-Control"))
+ }
+ }
+ if probeTransport(net.JoinHostPort(host, port)) != probeTransport(net.JoinHostPort(host, port)) {
+ t.Error("two transports for one address")
+ }
+ if ok, detail := probeAt(ctx, "127.0.0.2", "example.com", "/ready", "v1", "v42"); ok || !strings.Contains(detail, "answer") {
+ t.Errorf("an address that does not answer: %v %q", ok, detail)
+ }
+}
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index 7f2323a90..6c3b905bb 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -20,6 +20,7 @@ import (
"fmt"
"log"
"slices"
+ "strings"
"time"
"github.com/carlosframework/platform/internal/moves"
@@ -1072,6 +1073,19 @@ func (s *Server) selfServePreOfflineRefusal(row *SelfServeMove, err error, now t
return err
}
+// selfServeCutWriteRefused is a cut write after the start (the seal, the
+// flip, the open) that the record refused: the instance is offline, so the
+// abort names the step that stopped ("the move stopped at <step>: the
+// instance changed under it"). Any other error is returned.
+func (s *Server) selfServeCutWriteRefused(row *SelfServeMove, err error, step string, now time.Time) error {
+ var refusal *requestRefusal
+ if errors.As(err, &refusal) {
+ log.Printf("console: self-serve move %s: %v", row.ID, err)
+ return s.selfServeAbort(row, ssStoppedAt(step), now)
+ }
+ return err
+}
+
// selfServeRestageMut is the restage the staging, ready and restage
// phases write: staged_at moves to the proposed stamp.
func selfServeRestageMut(id string) func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
@@ -1543,8 +1557,13 @@ func (s *Server) selfServeReverting(ctx context.Context, t *leaseToken, row *Sel
if _, ref := s.offlineStepFor(ctx, d, mv, ev, now); ref != nil {
return nil // the abort's reports are not all in
}
- if left := selfServeTargetLeft(row, ev); left != "" && row.Warning == "" {
- if err := s.store.SetSelfServeWarning(row.ID, left); err != nil {
+ if left := selfServeTargetLeft(row, ev); left != "" && !strings.Contains(row.Warning, left) {
+ // Never dropped: appended to a warning already there.
+ w := left
+ if row.Warning != "" {
+ w = row.Warning + "; " + left
+ }
+ if err := s.store.SetSelfServeWarning(row.ID, w); err != nil {
return err
}
}
@@ -1726,7 +1745,7 @@ func (s *Server) selfServeStopping(ctx context.Context, t *leaseToken, row *Self
func(_ time.Time, rc provision.Receipt) func(*provision.Record) error {
return provision.OfflineSeal(row.ID, sealer, boxes, rc)
}); err != nil {
- return s.selfServePreOfflineRefusal(row, err, now)
+ return s.selfServeCutWriteRefused(row, err, factSeal, now)
}
if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
return err
@@ -1820,7 +1839,7 @@ func (s *Server) selfServeLanding(ctx context.Context, t *leaseToken, row *SelfS
func(_ time.Time, rc provision.Receipt) func(*provision.Record) error {
return provision.OfflineFlip(row.ID, rc)
}); err != nil {
- return s.selfServePreOfflineRefusal(row, err, now)
+ return s.selfServeCutWriteRefused(row, err, factFlip, now)
}
if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
return err
@@ -1895,7 +1914,7 @@ func (s *Server) selfServeLanded(ctx context.Context, t *leaseToken, row *SelfSe
switch {
case err == nil:
case errors.As(err, &refusal):
- return s.selfServePreOfflineRefusal(row, err, now)
+ return s.selfServeCutWriteRefused(row, err, factOpen, now)
case errors.Is(err, errRecordUnreadable):
return s.selfServeOpenUnconfirmable(row, now)
default:
diff --git a/internal/e2e/selfserve/cut_test.go b/internal/e2e/selfserve/cut_test.go
index b795db25b..5e00d65c7 100644
--- a/internal/e2e/selfserve/cut_test.go
+++ b/internal/e2e/selfserve/cut_test.go
@@ -404,6 +404,11 @@ func TestAbortIsInitiatedNotCompleted(t *testing.T) {
if !ok || sealer.Phase != moves.OfflinePhaseReverted || !sealer.TombstoneLifted || sealer.TombstoneLiftedAt.IsZero() {
t.Fatalf("the sealer's report %+v", sealer)
}
+ // The source reopened only once the real sealer had lifted the
+ // tombstone (its durable lift time).
+ if r.SourceReopenedAt.Before(sealer.TombstoneLiftedAt) {
+ t.Fatalf("source reopened at %v, before the lift at %v", r.SourceReopenedAt, sealer.TombstoneLiftedAt)
+ }
if tgt, _ := w.report(r.ID, moves.OfflineRoleTarget, host.E2ETargetBox); tgt.Phase == moves.OfflinePhaseReverted {
t.Fatal("the target reported its revert while down")
}
diff --git a/internal/e2e/selfserve/job_boundary_test.go b/internal/e2e/selfserve/job_boundary_test.go
index f4a9c03e8..5e328fe43 100644
--- a/internal/e2e/selfserve/job_boundary_test.go
+++ b/internal/e2e/selfserve/job_boundary_test.go
@@ -308,9 +308,11 @@ func everyBoundary(t *testing.T, b jobBoundary, f jobFault, second bool) {
}
}
final := w.row(row.ID)
- if !confirmed && !(final.Request.Nonce == nonce && final.Request.Confirmed) && !strings.HasPrefix(b.name, "abort") && !strings.HasPrefix(b.name, "finish") {
+ if !confirmed && !(final.Request.Nonce == nonce && final.Request.Confirmed) && !strings.HasPrefix(b.name, "abort") && b.name != "finish" && b.name != "finish-done" {
// A clearing request's confirmation and its consumption by the
- // finaliser can share one tick; every other must be seen.
+ // finaliser can share one tick (the abort's, the reverting
+ // finish's, gate 2's finish); every other must be seen, gate 1's
+ // included.
t.Fatalf("request %s was never confirmed (final %s)", nonce, describe(final))
}
if len(unannounced) > 0 {