amadan
host: PageOperator, the paging primitive for moves (§13)
The webhook and SNS delivery moves to internal/alerting, which the agent
and the console can both import (K11). host's alertSinks is a thin user
of it: alerts, refusals and their failure counts are unchanged, and the
existing alertsink tests pass as they were.
alerting.Page(ctx, level, subject, body) pages on the armed sinks: the
webhook gets {"kind":"page",...}, SNS the subject (made to fit) and body
with the level as a message attribute. With no sink it logs once and
returns nil. The metrics loop arms the agent's sinks; host.PageOperator
wraps alerting.Page.
moveCfg.Page is PageOperator in armMovesPass and nil in tests. The box
arms page an incident wherever a MoveMarkStuck lands: the target's
return (movesrollback.go), the source's recovery and its dirty return's
install (movesrecover.go) and the revert (movesrevert.go). The subject
and body are MV-N313 and MV-N314, pending; they name the host, the move,
the box, its role and the recorded reason, and no path or secret.
The B8 census digests of recoverRun.stuck and rollbackRun.stuck change
for the added page call after the write; the write is unchanged.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
15 files changed,
+775
−97
approved_moves_copy_test.go+2 −0docs/superpowers/plans/2026-10-06-moves-phase1-copy.md+6 −1internal/alerting/alerting.go+245 −0internal/alerting/alerting_test.go+208 −0internal/host/alertsink.go+52 −77internal/host/alertsink_page_test.go+91 −0internal/host/metrics_loop.go+4 −15internal/host/movespass.go+8 −0internal/host/movesrecover.go+2 −0internal/host/movesrecover_test.go+1 −0internal/host/movesrevert.go+1 −0internal/host/movesrollback.go+36 −2internal/host/movesrollback_test.go+2 −0internal/host/movesstuckpage_test.go+115 −0internal/provision/plan9_b8_census_test.go+2 −2
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go| index 4803a1430..1a34e65b4 100644 |
| --- a/approved_moves_copy_test.go |
| +++ b/approved_moves_copy_test.go |
| @@ -73,6 +73,7 @@ var movesCopyIDs = []string{ |
| "MV-N182", "MV-N183", "MV-N184", "MV-N185", "MV-N186", |
| "MV-N300", "MV-N301", "MV-N302", "MV-N303", "MV-N304", "MV-N305", "MV-N306", |
| "MV-N307", "MV-N308", "MV-N309", "MV-N310", "MV-N311", "MV-N312", |
| + "MV-N313", "MV-N314", |
| } |
| // copyRow is one row of the record. The File cell is the file the sentence is |
| @@ -725,6 +726,7 @@ var movesReturnedSentences = map[string]map[string]bool{ |
| "s1OverCap": true, "s1LongPath": true, "s1NoAnswer": true, "s1ManyBoxes": true, "oneBoxRefusal": true}, |
| "internal/host/movesreport.go": {"t1Below": true, "t3Differs": true, "t3Shadowed": true, "t4Failed": true, |
| "t6NoRoom": true, "t8Count": true, "t12NotQuiet": true, "replicaReason": true}, |
| + "internal/host/movesrollback.go": {"stuckPage": true}, |
| } |
| // movesLiteralFuncs are functions whose every string literal is a sentence |
diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md| index b56d80ec7..c908fef05 100644 |
| --- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md |
| +++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md |
| @@ -263,7 +263,12 @@ Columns: id, file, text, status. |
| ## Plan 5 |
| -(appended by Plan 5 Task 8) |
| +(each task's rows in the task that writes them; the rest appended by Plan 5 Task 8. Task 0's stuck page continues Plan 4's MV-N3xx block, ruling K11) |
| + |
| +| id | file | text | status | |
| +|---|---|---|---| |
| +| MV-N313 | internal/host/movesrollback.go | carlos move stuck: <host> | pending | |
| +| MV-N314 | internal/host/movesrollback.go | The move <id> of <host> stopped for an operator on <box>, its <role>: <reason>. Run carlos move resolve --host <host> --retry once the cause is fixed. | pending | |
| ## Plan 6 |
diff --git a/internal/alerting/alerting.go b/internal/alerting/alerting.go| new file mode 100644 |
| index 000000000..782cc052b |
| --- /dev/null |
| +++ b/internal/alerting/alerting.go |
| @@ -0,0 +1,245 @@ |
| +// Package alerting is the operator alert delivery the agent and the console |
| +// can share: a webhook, an SNS topic, either, both or neither (Sinks), and |
| +// Page, the paging primitive the moves code calls (moves spec §13). |
| +// |
| +// Delivery is best effort. A sink that fails is counted through its failure |
| +// hook and its error returned; the other sink still runs. Nothing here |
| +// retries beyond what the AWS SDK does inside one publish. |
| +package alerting |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "net/http" |
| + "regexp" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "time" |
| + |
| + "github.com/aws/aws-sdk-go-v2/aws" |
| + "github.com/aws/aws-sdk-go-v2/aws/arn" |
| + awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| + "github.com/aws/aws-sdk-go-v2/service/sns" |
| + snstypes "github.com/aws/aws-sdk-go-v2/service/sns/types" |
| +) |
| + |
| +// Page levels (moves spec §13): an incident wants someone now, a warning |
| +// wants someone today. |
| +const ( |
| + LevelIncident = "incident" |
| + LevelWarning = "warning" |
| +) |
| + |
| +// Publisher is the one SNS call this package makes. |
| +type Publisher interface { |
| + Publish(context.Context, *sns.PublishInput, ...func(*sns.Options)) (*sns.PublishOutput, error) |
| +} |
| + |
| +// PublishTimeout bounds one SNS publish, so a black-holed endpoint cannot |
| +// hold the caller for the SDK's own default retries. |
| +const PublishTimeout = 10 * time.Second |
| + |
| +// subjectLimit is SNS's: a subject is under 100 characters. |
| +const subjectLimit = 100 |
| + |
| +// Sinks is one process's delivery. An empty WebhookURL or a nil Post |
| +// switches the webhook off; an empty TopicARN or a nil Pub switches SNS off. |
| +// The failure hooks count a failed send; nil counts nothing. |
| +type Sinks struct { |
| + WebhookURL string |
| + Post func(url string, body []byte) error |
| + TopicARN string |
| + Pub Publisher |
| + |
| + WebhookFailed func() |
| + SNSFailed func() |
| +} |
| + |
| +// HasWebhook and HasSNS say whether that sink is on. Safe on nil. |
| +func (s *Sinks) HasWebhook() bool { return s != nil && s.WebhookURL != "" && s.Post != nil } |
| +func (s *Sinks) HasSNS() bool { return s != nil && s.TopicARN != "" && s.Pub != nil } |
| + |
| +// PostWebhook posts body to the webhook. A failure is counted and returned. |
| +// The caller checks HasWebhook first. |
| +func (s *Sinks) PostWebhook(body []byte) error { |
| + if err := s.Post(s.WebhookURL, body); err != nil { |
| + if s.WebhookFailed != nil { |
| + s.WebhookFailed() |
| + } |
| + return err |
| + } |
| + return nil |
| +} |
| + |
| +// PublishSNS publishes a subject and message to the topic, bounded by |
| +// PublishTimeout. A failure is counted and returned. The caller checks |
| +// HasSNS first. |
| +func (s *Sinks) PublishSNS(ctx context.Context, subject, message string) error { |
| + return s.publish(ctx, &sns.PublishInput{TopicArn: &s.TopicARN, Subject: &subject, Message: &message}) |
| +} |
| + |
| +func (s *Sinks) publish(ctx context.Context, in *sns.PublishInput) error { |
| + pctx, cancel := context.WithTimeout(ctx, PublishTimeout) |
| + defer cancel() |
| + if _, err := s.Pub.Publish(pctx, in); err != nil { |
| + if s.SNSFailed != nil { |
| + s.SNSFailed() |
| + } |
| + return err |
| + } |
| + return nil |
| +} |
| + |
| +// pagePayload is a page on the webhook. |
| +type pagePayload struct { |
| + Kind string `json:"kind"` |
| + Level string `json:"level"` |
| + Subject string `json:"subject"` |
| + Body string `json:"body"` |
| +} |
| + |
| +// Page sends one page to every configured sink: the webhook gets |
| +// {"kind":"page","level","subject","body"}, SNS the subject (made to fit |
| +// SNS's rules) and the body, with the level as a message attribute so a |
| +// subscription can filter on it. The caller's subject and body are the |
| +// whole message: no URL, ARN or credential is added here, and the caller |
| +// must not put a secret or a tenant path in either. With no sink it logs |
| +// once (until the next Arm) and returns nil. A failed sink is counted, and |
| +// every failure is returned joined. |
| +func (s *Sinks) Page(ctx context.Context, level, subject, body string) error { |
| + if !s.HasWebhook() && !s.HasSNS() { |
| + if !noSinkLogged.Swap(true) { |
| + log.Printf("alerting: page not delivered, no sink configured (set CARLOS_ALERT_SNS_TOPIC_ARN or CARLOS_ALERT_WEBHOOK): %s", fitSubject(subject)) |
| + } |
| + return nil |
| + } |
| + var errs []error |
| + if s.HasWebhook() { |
| + b, err := json.Marshal(pagePayload{Kind: "page", Level: level, Subject: subject, Body: body}) |
| + if err == nil { |
| + err = s.PostWebhook(b) |
| + } |
| + if err != nil { |
| + errs = append(errs, fmt.Errorf("webhook: %w", err)) |
| + } |
| + } |
| + if s.HasSNS() { |
| + subj := fitSubject(subject) |
| + in := &sns.PublishInput{TopicArn: &s.TopicARN, Subject: &subj, Message: &body, |
| + MessageAttributes: map[string]snstypes.MessageAttributeValue{"level": {DataType: aws.String("String"), StringValue: aws.String(level)}}} |
| + if err := s.publish(ctx, in); err != nil { |
| + errs = append(errs, fmt.Errorf("sns: %w", err)) |
| + } |
| + } |
| + return errors.Join(errs...) |
| +} |
| + |
| +// fitSubject makes a subject SNS accepts: printable ASCII, no line break, |
| +// under 100 characters. A line break becomes a space and anything else |
| +// outside printable ASCII a question mark. |
| +func fitSubject(s string) string { |
| + out := make([]byte, 0, len(s)) |
| + for _, r := range s { |
| + switch { |
| + case r == '\n' || r == '\r' || r == '\t': |
| + if n := len(out); n == 0 || out[n-1] != ' ' { |
| + out = append(out, ' ') |
| + } |
| + case r < 0x20 || r > 0x7e: |
| + out = append(out, '?') |
| + default: |
| + out = append(out, byte(r)) |
| + } |
| + if len(out) >= subjectLimit-1 { |
| + break |
| + } |
| + } |
| + return string(out) |
| +} |
| + |
| +// The armed sinks: the process's own, set once its sinks are built (Arm). |
| +var ( |
| + armedMu sync.RWMutex |
| + armed *Sinks |
| + noSinkLogged atomic.Bool |
| +) |
| + |
| +// Arm makes s the sinks Page delivers on. nil means none. Arming starts a |
| +// fresh "no sink" notice. |
| +func Arm(s *Sinks) { |
| + armedMu.Lock() |
| + armed = s |
| + armedMu.Unlock() |
| + noSinkLogged.Store(false) |
| +} |
| + |
| +// Page pages an operator on the armed sinks (Sinks.Page). |
| +func Page(ctx context.Context, level, subject, body string) error { |
| + armedMu.RLock() |
| + s := armed |
| + armedMu.RUnlock() |
| + return s.Page(ctx, level, subject, body) |
| +} |
| + |
| +// PostJSON is the production webhook post: a 5 second POST with a JSON |
| +// content type. Any non-2xx status is an error. |
| +func PostJSON(url string, body []byte) error { |
| + client := &http.Client{Timeout: 5 * time.Second} |
| + resp, err := client.Post(url, "application/json", bytes.NewReader(body)) |
| + if err != nil { |
| + return err |
| + } |
| + defer resp.Body.Close() |
| + if resp.StatusCode < 200 || resp.StatusCode >= 300 { |
| + return fmt.Errorf("host: webhook: non-2xx status %s", resp.Status) |
| + } |
| + return nil |
| +} |
| + |
| +// topicARNPattern is the whole shape of a standard SNS topic ARN: partition, |
| +// region, a 12 digit account, and a topic name that is only letters, digits, |
| +// hyphen and underscore. The name has no colon, so a subscription ARN |
| +// (topic ARN plus ":<subscription-id>") does not match, and no dot, so FIFO |
| +// topics (which need a message group id this sink does not send) do not |
| +// either. |
| +var topicARNPattern = regexp.MustCompile(`^arn:aws(-cn|-us-gov)?:sns:([a-z0-9-]+):[0-9]{12}:[A-Za-z0-9_-]{1,256}$`) |
| + |
| +// ParseTopicARN checks a CARLOS_ALERT_SNS_TOPIC_ARN value and returns the |
| +// region the topic lives in. SNS is regional and a topic is reached only |
| +// through its own region's endpoint, so the region is read from the ARN and |
| +// not from AWS_REGION (host.env pins that to the deployment bucket's region |
| +// on every flagship box, which would aim three of the four at the wrong |
| +// place). |
| +func ParseTopicARN(s string) (region string, err error) { |
| + s = strings.TrimSpace(s) |
| + if s == "" { |
| + return "", errors.New("empty") |
| + } |
| + a, err := arn.Parse(s) |
| + if err != nil { |
| + return "", err |
| + } |
| + if a.Service != "sns" { |
| + return "", fmt.Errorf("service %q, want sns", a.Service) |
| + } |
| + if !topicARNPattern.MatchString(s) { |
| + return "", errors.New("not a topic arn (want arn:aws:sns:<region>:<12 digit account>:<topic name>)") |
| + } |
| + return a.Region, nil |
| +} |
| + |
| +// NewSNSClient is the production publisher: the default credential chain |
| +// (a box's instance role, the console's task role), aimed at the topic's |
| +// own region. |
| +func NewSNSClient(ctx context.Context, region string) (Publisher, error) { |
| + cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region)) |
| + if err != nil { |
| + return nil, err |
| + } |
| + return sns.NewFromConfig(cfg), nil |
| +} |
diff --git a/internal/alerting/alerting_test.go b/internal/alerting/alerting_test.go| new file mode 100644 |
| index 000000000..90df95149 |
| --- /dev/null |
| +++ b/internal/alerting/alerting_test.go |
| @@ -0,0 +1,208 @@ |
| +package alerting |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "log" |
| + "strings" |
| + "testing" |
| + |
| + "github.com/aws/aws-sdk-go-v2/service/sns" |
| +) |
| + |
| +// fakeSNS records every Publish and can be told to fail. |
| +type fakeSNS struct { |
| + in []*sns.PublishInput |
| + err error |
| +} |
| + |
| +func (f *fakeSNS) Publish(_ context.Context, in *sns.PublishInput, _ ...func(*sns.Options)) (*sns.PublishOutput, error) { |
| + f.in = append(f.in, in) |
| + if f.err != nil { |
| + return nil, f.err |
| + } |
| + return &sns.PublishOutput{}, nil |
| +} |
| + |
| +const testTopicARN = "arn:aws:sns:eu-west-1:271376211898:carlos-flagship-ie-alerts" |
| + |
| +type posted struct { |
| + url string |
| + body []byte |
| +} |
| + |
| +func poster(into *[]posted, err error) func(string, []byte) error { |
| + return func(url string, body []byte) error { |
| + *into = append(*into, posted{url, body}) |
| + return err |
| + } |
| +} |
| + |
| +// captureLog sends the standard logger to a buffer for the test. |
| +func captureLog(t *testing.T) *bytes.Buffer { |
| + t.Helper() |
| + var buf bytes.Buffer |
| + prev, flags := log.Writer(), log.Flags() |
| + log.SetOutput(&buf) |
| + log.SetFlags(0) |
| + t.Cleanup(func() { log.SetOutput(prev); log.SetFlags(flags) }) |
| + return &buf |
| +} |
| + |
| +func TestPageDeliversOnBothSinks(t *testing.T) { |
| + var posts []posted |
| + pub := &fakeSNS{} |
| + s := &Sinks{WebhookURL: "http://example.test/hook", Post: poster(&posts, nil), TopicARN: testTopicARN, Pub: pub} |
| + if err := s.Page(context.Background(), LevelIncident, "carlos move stuck: shop.example", "the body.\nsecond line."); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(posts) != 1 || posts[0].url != "http://example.test/hook" { |
| + t.Fatalf("webhook posts: %+v", posts) |
| + } |
| + var got map[string]string |
| + if err := json.Unmarshal(posts[0].body, &got); err != nil { |
| + t.Fatal(err) |
| + } |
| + want := map[string]string{"kind": "page", "level": "incident", "subject": "carlos move stuck: shop.example", "body": "the body.\nsecond line."} |
| + if len(got) != len(want) { |
| + t.Fatalf("webhook payload %s", posts[0].body) |
| + } |
| + for k, v := range want { |
| + if got[k] != v { |
| + t.Fatalf("webhook payload %s: %s = %q, want %q", posts[0].body, k, got[k], v) |
| + } |
| + } |
| + if len(pub.in) != 1 { |
| + t.Fatalf("publishes = %d, want 1", len(pub.in)) |
| + } |
| + in := pub.in[0] |
| + if *in.TopicArn != testTopicARN || *in.Subject != "carlos move stuck: shop.example" || *in.Message != "the body.\nsecond line." { |
| + t.Fatalf("publish: %q %q %q", *in.TopicArn, *in.Subject, *in.Message) |
| + } |
| + if a, ok := in.MessageAttributes["level"]; !ok || a.StringValue == nil || *a.StringValue != "incident" || *a.DataType != "String" { |
| + t.Fatalf("the level attribute: %+v", in.MessageAttributes) |
| + } |
| +} |
| + |
| +// SNS takes a subject of under 100 printable ASCII characters with no line |
| +// break; a page's subject is made to fit rather than refused. |
| +func TestPageSubjectFitsSNS(t *testing.T) { |
| + pub := &fakeSNS{} |
| + s := &Sinks{TopicARN: testTopicARN, Pub: pub} |
| + long := "carlos move stuck:\n" + strings.Repeat("x", 120) + "é" |
| + if err := s.Page(context.Background(), LevelWarning, long, "body"); err != nil { |
| + t.Fatal(err) |
| + } |
| + got := *pub.in[0].Subject |
| + if len(got) >= 100 || strings.ContainsAny(got, "\r\n") || !strings.HasPrefix(got, "carlos move stuck: x") { |
| + t.Fatalf("subject %q (%d)", got, len(got)) |
| + } |
| + for _, r := range got { |
| + if r < 0x20 || r > 0x7e { |
| + t.Fatalf("subject %q has %q", got, r) |
| + } |
| + } |
| +} |
| + |
| +// A failing sink is counted and returned, and never stops the other one. |
| +func TestPageFailureIsCountedAndReturned(t *testing.T) { |
| + var posts []posted |
| + pub := &fakeSNS{err: errors.New("AccessDenied")} |
| + var webhookFailed, snsFailed int |
| + s := &Sinks{WebhookURL: "http://example.test/hook", Post: poster(&posts, errors.New("boom")), TopicARN: testTopicARN, Pub: pub, |
| + WebhookFailed: func() { webhookFailed++ }, SNSFailed: func() { snsFailed++ }} |
| + err := s.Page(context.Background(), LevelIncident, "s", "b") |
| + if err == nil || !strings.Contains(err.Error(), "boom") || !strings.Contains(err.Error(), "AccessDenied") { |
| + t.Fatalf("err = %v, want both failures", err) |
| + } |
| + if len(posts) != 1 || len(pub.in) != 1 || webhookFailed != 1 || snsFailed != 1 { |
| + t.Fatalf("posts %d publishes %d failures %d/%d", len(posts), len(pub.in), webhookFailed, snsFailed) |
| + } |
| +} |
| + |
| +// With no sink configured a page is not an error: it says so once in the |
| +// journal and returns nil, as the alert delivery does. |
| +func TestPageWithNoSinksLogsOnce(t *testing.T) { |
| + for _, c := range []struct { |
| + name string |
| + sinks *Sinks |
| + }{{"nil", nil}, {"empty", &Sinks{}}, {"a webhook with no poster", &Sinks{WebhookURL: "http://example.test/hook"}}} { |
| + t.Run(c.name, func(t *testing.T) { |
| + buf := captureLog(t) |
| + Arm(c.sinks) |
| + t.Cleanup(func() { Arm(nil) }) |
| + for range 3 { |
| + if err := Page(context.Background(), LevelIncident, "s", "b"); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if n := strings.Count(buf.String(), "no sink configured"); n != 1 { |
| + t.Fatalf("logged %d times, want once:\n%s", n, buf) |
| + } |
| + // Arming again starts a fresh notice. |
| + Arm(c.sinks) |
| + if err := Page(context.Background(), LevelIncident, "s", "b"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if n := strings.Count(buf.String(), "no sink configured"); n != 2 { |
| + t.Fatalf("after re-arming: logged %d times, want twice:\n%s", n, buf) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestPageUsesTheArmedSinks(t *testing.T) { |
| + var posts []posted |
| + Arm(&Sinks{WebhookURL: "http://example.test/hook", Post: poster(&posts, nil)}) |
| + t.Cleanup(func() { Arm(nil) }) |
| + if err := Page(context.Background(), LevelWarning, "subject", "body"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(posts) != 1 || !strings.Contains(string(posts[0].body), `"subject":"subject"`) { |
| + t.Fatalf("posts: %+v", posts) |
| + } |
| +} |
| + |
| +// The delivery the alert sinks share: a webhook post and an SNS publish, |
| +// each counted on failure. |
| +func TestSinksPostAndPublish(t *testing.T) { |
| + var posts []posted |
| + pub := &fakeSNS{} |
| + var webhookFailed int |
| + s := &Sinks{WebhookURL: "http://example.test/hook", Post: poster(&posts, errors.New("boom")), TopicARN: testTopicARN, Pub: pub, |
| + WebhookFailed: func() { webhookFailed++ }} |
| + if !s.HasWebhook() || !s.HasSNS() { |
| + t.Fatal("both sinks are configured") |
| + } |
| + if err := s.PostWebhook([]byte(`{}`)); err == nil || webhookFailed != 1 { |
| + t.Fatalf("post: %v, %d failures", err, webhookFailed) |
| + } |
| + if err := s.PublishSNS(context.Background(), "subj", "msg"); err != nil || len(pub.in) != 1 || *pub.in[0].Subject != "subj" || pub.in[0].MessageAttributes != nil { |
| + t.Fatalf("publish: %v %+v", err, pub.in) |
| + } |
| + var none *Sinks |
| + if none.HasWebhook() || none.HasSNS() { |
| + t.Fatal("a nil sink set has no sinks") |
| + } |
| +} |
| + |
| +func TestParseTopicARN(t *testing.T) { |
| + for _, c := range []struct { |
| + in, region string |
| + ok bool |
| + }{ |
| + {testTopicARN, "eu-west-1", true}, |
| + {" " + testTopicARN + " ", "eu-west-1", true}, |
| + {"", "", false}, |
| + {testTopicARN + ":sub-id", "", false}, |
| + {"arn:aws:sqs:eu-west-1:271376211898:q", "", false}, |
| + {"arn:aws:sns:eu-west-1:271376211898:topic.fifo", "", false}, |
| + } { |
| + region, err := ParseTopicARN(c.in) |
| + if (err == nil) != c.ok || region != c.region { |
| + t.Errorf("ParseTopicARN(%q) = %q, %v", c.in, region, err) |
| + } |
| + } |
| +} |
diff --git a/internal/host/alertsink.go b/internal/host/alertsink.go| index bd5c081b9..ea204da0e 100644 |
| --- a/internal/host/alertsink.go |
| +++ b/internal/host/alertsink.go |
| @@ -3,19 +3,14 @@ package host |
| import ( |
| "context" |
| "encoding/json" |
| - "errors" |
| "fmt" |
| "log" |
| "math" |
| - "regexp" |
| "strings" |
| "sync/atomic" |
| "time" |
| - "github.com/aws/aws-sdk-go-v2/aws/arn" |
| - awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| - "github.com/aws/aws-sdk-go-v2/service/sns" |
| - |
| + "github.com/carlosframework/platform/internal/alerting" |
| "github.com/carlosframework/platform/internal/metrics" |
| "github.com/carlosframework/platform/internal/release" |
| "github.com/carlosframework/platform/internal/status" |
| @@ -44,15 +39,14 @@ import ( |
| // write in alertTick repeats the whole transition next tick, delivery |
| // included.) The count is what makes a lost send visible, through |
| // status.AlertDelivery in the box snapshot. |
| +// |
| +// The delivery itself (the webhook post, the SNS publish and its timeout) |
| +// is internal/alerting's, shared with the console; this file formats the |
| +// box's alerts and counts their failures. The same sinks carry the |
| +// operator pages of the moves code (PageOperator, moves spec §13). |
| -// snsPublisher is the one SNS call this file makes. |
| -type snsPublisher interface { |
| - Publish(context.Context, *sns.PublishInput, ...func(*sns.Options)) (*sns.PublishOutput, error) |
| -} |
| - |
| -// alertPublishTimeout bounds one SNS publish, so a black-holed endpoint |
| -// cannot hold the 60 second sampler tick for the SDK's own default retries. |
| -const alertPublishTimeout = 10 * time.Second |
| +// snsPublisher is the one SNS call the sinks make. |
| +type snsPublisher = alerting.Publisher |
| // alertCounters counts failed sends since the process started, and holds the |
| // one fact the status loop cannot derive from the env: whether the SNS client |
| @@ -76,13 +70,11 @@ const ( |
| // shared state, so this is the seam between them. |
| var processAlertCounters alertCounters |
| -// alertSinks is the configured delivery for one agent. |
| +// alertSinks is the configured delivery for one agent: alerting's sinks, |
| +// counting their failures in counts. |
| type alertSinks struct { |
| - webhookURL string |
| - post func(url string, body []byte) error |
| - topicARN string |
| - pub snsPublisher |
| - counts *alertCounters |
| + sinks alerting.Sinks |
| + counts *alertCounters |
| } |
| // newAlertSinks builds a sink set. An empty webhookURL or topicARN switches |
| @@ -92,7 +84,8 @@ func newAlertSinks(webhookURL string, post func(url string, body []byte) error, |
| if counts == nil { |
| counts = &processAlertCounters |
| } |
| - return &alertSinks{webhookURL: webhookURL, post: post, topicARN: topicARN, pub: pub, counts: counts} |
| + return &alertSinks{counts: counts, sinks: alerting.Sinks{WebhookURL: webhookURL, Post: post, TopicARN: topicARN, Pub: pub, |
| + WebhookFailed: func() { counts.webhook.Add(1) }, SNSFailed: func() { counts.sns.Add(1) }}} |
| } |
| // deliver sends one transition to every configured sink. Safe on a nil |
| @@ -101,38 +94,56 @@ func (a *alertSinks) deliver(ctx context.Context, rules metrics.Rules, at time.T |
| if a == nil { |
| return |
| } |
| - if a.webhookURL != "" && a.post != nil { |
| + if a.sinks.HasWebhook() { |
| evBody, err := json.Marshal(ev) |
| if err != nil { |
| log.Printf("host: alert event %s: marshal: %v", ev.Rule, err) |
| - } else if err := a.post(a.webhookURL, evBody); err != nil { |
| - a.counts.webhook.Add(1) |
| + } else if err := a.sinks.PostWebhook(evBody); err != nil { |
| log.Printf("host: alert webhook post (rule %s): %v", ev.Rule, err) |
| } |
| } |
| - if a.topicARN != "" && a.pub != nil { |
| + if a.sinks.HasSNS() { |
| subject, message := formatAlert(rules, at, ev) |
| - pctx, cancel := context.WithTimeout(ctx, alertPublishTimeout) |
| - defer cancel() |
| - _, err := a.pub.Publish(pctx, &sns.PublishInput{ |
| - TopicArn: &a.topicARN, |
| - Subject: &subject, |
| - Message: &message, |
| - }) |
| - if err != nil { |
| - a.counts.sns.Add(1) |
| + if err := a.sinks.PublishSNS(ctx, subject, message); err != nil { |
| log.Printf("host: alert sns publish (rule %s): %v", ev.Rule, err) |
| } |
| } |
| } |
| +// PageOperator pages an operator through this agent's sinks: the same |
| +// webhook and SNS topic deliver uses, its failures counted with theirs |
| +// (alerting.Sinks.Page). Safe on a nil receiver, which means "no sinks": |
| +// nil, and one line in the journal. |
| +func (a *alertSinks) PageOperator(ctx context.Context, level, subject, body string) error { |
| + if a == nil { |
| + return (*alerting.Sinks)(nil).Page(ctx, level, subject, body) |
| + } |
| + return a.sinks.Page(ctx, level, subject, body) |
| +} |
| + |
| +// PageOperator pages an operator on the sinks the metrics loop armed |
| +// (armPaging): the moves code's paging primitive (moves spec §13). Before |
| +// they are armed, or with none configured, it returns nil and says so once. |
| +func PageOperator(ctx context.Context, level, subject, body string) error { |
| + return alerting.Page(ctx, level, subject, body) |
| +} |
| + |
| +// armPaging makes a the sinks PageOperator pages on. nil means none. |
| +func armPaging(a *alertSinks) { |
| + if a == nil { |
| + alerting.Arm(nil) |
| + return |
| + } |
| + alerting.Arm(&a.sinks) |
| +} |
| + |
| // deliverRelease sends a D1 refusal to every configured sink. The webhook |
| // body is the record with kind "release_refused"; SNS gets a sentence. |
| func (a *alertSinks) deliverRelease(ctx context.Context, r release.Refusal) { |
| if a == nil { |
| return |
| } |
| - if a.webhookURL != "" && a.post != nil { |
| + if a.sinks.HasWebhook() { |
| body, err := json.Marshal(struct { |
| Kind string `json:"kind"` |
| // The record's own Kind is shadowed by the envelope's "kind" |
| @@ -141,18 +152,14 @@ func (a *alertSinks) deliverRelease(ctx context.Context, r release.Refusal) { |
| release.Refusal |
| }{"release_refused", r.Kind, r}) |
| if err == nil { |
| - if err := a.post(a.webhookURL, body); err != nil { |
| - a.counts.webhook.Add(1) |
| + if err := a.sinks.PostWebhook(body); err != nil { |
| log.Printf("host: alert webhook post (release refusal %s): %v", r.Host, err) |
| } |
| } |
| } |
| - if a.topicARN != "" && a.pub != nil { |
| + if a.sinks.HasSNS() { |
| subject, msg := formatReleaseAlert(r) |
| - pctx, cancel := context.WithTimeout(ctx, alertPublishTimeout) |
| - defer cancel() |
| - if _, err := a.pub.Publish(pctx, &sns.PublishInput{TopicArn: &a.topicARN, Subject: &subject, Message: &msg}); err != nil { |
| - a.counts.sns.Add(1) |
| + if err := a.sinks.PublishSNS(ctx, subject, msg); err != nil { |
| log.Printf("host: alert sns publish (release refusal %s): %v", r.Host, err) |
| } |
| } |
| @@ -215,37 +222,9 @@ func recoveryError(r release.Refusal) string { |
| return "outcome not recorded" |
| } |
| -// topicARNPattern is the whole shape of a standard SNS topic ARN: partition, |
| -// region, a 12 digit account, and a topic name that is only letters, digits, |
| -// hyphen and underscore. The name has no colon, so a subscription ARN |
| -// (topic ARN plus ":<subscription-id>") does not match, and no dot, so FIFO |
| -// topics (which need a message group id this sink does not send) do not |
| -// either. |
| -var topicARNPattern = regexp.MustCompile(`^arn:aws(-cn|-us-gov)?:sns:([a-z0-9-]+):[0-9]{12}:[A-Za-z0-9_-]{1,256}$`) |
| - |
| // parseAlertTopicARN checks CARLOS_ALERT_SNS_TOPIC_ARN and returns the |
| -// region the topic lives in. SNS is regional and a topic is reached only |
| -// through its own region's endpoint, so the region is read from the ARN and |
| -// not from AWS_REGION (host.env pins that to the deployment bucket's region |
| -// on every flagship box, which would aim three of the four at the wrong |
| -// place). |
| -func parseAlertTopicARN(s string) (region string, err error) { |
| - s = strings.TrimSpace(s) |
| - if s == "" { |
| - return "", errors.New("empty") |
| - } |
| - a, err := arn.Parse(s) |
| - if err != nil { |
| - return "", err |
| - } |
| - if a.Service != "sns" { |
| - return "", fmt.Errorf("service %q, want sns", a.Service) |
| - } |
| - if !topicARNPattern.MatchString(s) { |
| - return "", errors.New("not a topic arn (want arn:aws:sns:<region>:<12 digit account>:<topic name>)") |
| - } |
| - return a.Region, nil |
| -} |
| +// region the topic lives in (alerting.ParseTopicARN). |
| +func parseAlertTopicARN(s string) (region string, err error) { return alerting.ParseTopicARN(s) } |
| // alertDeliveryProbe is the box's answer to "do my alert rules have anywhere |
| // to go". The webhook and the ARN's shape come from the same env |
| @@ -283,11 +262,7 @@ type snsClientFactory func(ctx context.Context, region string) (snsPublisher, er |
| // newSNSClient is the production factory: the default credential chain (the |
| // box's instance role), aimed at the topic's own region. |
| func newSNSClient(ctx context.Context, region string) (snsPublisher, error) { |
| - cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region)) |
| - if err != nil { |
| - return nil, err |
| - } |
| - return sns.NewFromConfig(cfg), nil |
| + return alerting.NewSNSClient(ctx, region) |
| } |
| // buildAlertSinks reads the sink env and builds the sinks. A bad topic ARN |
diff --git a/internal/host/alertsink_page_test.go b/internal/host/alertsink_page_test.go| new file mode 100644 |
| index 000000000..97f4fe123 |
| --- /dev/null |
| +++ b/internal/host/alertsink_page_test.go |
| @@ -0,0 +1,91 @@ |
| +package host |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "log" |
| + "strings" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/alerting" |
| +) |
| + |
| +// TestPageOperatorUsesTheArmedSinks: the paging primitive goes out through |
| +// the same webhook (and SNS topic) the agent's alerts use, its failures are |
| +// counted where the alert failures are, and the package function pages on |
| +// whatever the metrics loop armed. With no sinks it returns nil and says so |
| +// once. |
| +func TestPageOperatorUsesTheArmedSinks(t *testing.T) { |
| + type page struct{ Kind, Level, Subject, Body string } |
| + var pages []page |
| + var fail error |
| + post := func(url string, body []byte) error { |
| + var p page |
| + if err := json.Unmarshal(body, &p); err != nil { |
| + t.Fatalf("webhook body %s: %v", body, err) |
| + } |
| + if url != "http://example.test/hook" { |
| + t.Fatalf("posted to %s", url) |
| + } |
| + pages = append(pages, p) |
| + return fail |
| + } |
| + counts := &alertCounters{} |
| + pub := &fakeSNS{} |
| + sinks := newAlertSinks("http://example.test/hook", post, testTopicARN, pub, counts) |
| + ctx := context.Background() |
| + |
| + if err := sinks.PageOperator(ctx, alerting.LevelIncident, "carlos move stuck: shop.example", "the body"); err != nil { |
| + t.Fatal(err) |
| + } |
| + want := page{"page", "incident", "carlos move stuck: shop.example", "the body"} |
| + if len(pages) != 1 || pages[0] != want { |
| + t.Fatalf("pages %+v, want %+v", pages, want) |
| + } |
| + if len(pub.in) != 1 || deref(pub.in[0].Subject) != want.Subject || deref(pub.in[0].Message) != want.Body { |
| + t.Fatalf("sns: %+v", pub.in) |
| + } |
| + |
| + // The package function pages on the armed sinks. |
| + armPaging(sinks) |
| + t.Cleanup(func() { armPaging(nil) }) |
| + if err := PageOperator(ctx, alerting.LevelWarning, "second", "body two"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(pages) != 2 || pages[1].Subject != "second" || pages[1].Level != "warning" { |
| + t.Fatalf("pages %+v", pages) |
| + } |
| + |
| + // A failed send is returned and counted with the alert failures. |
| + fail = errors.New("boom") |
| + if err := PageOperator(ctx, alerting.LevelIncident, "third", "body"); err == nil { |
| + t.Fatal("a failed webhook post returned nil") |
| + } |
| + if counts.webhook.Load() != 1 || counts.sns.Load() != 0 { |
| + t.Fatalf("failures webhook=%d sns=%d", counts.webhook.Load(), counts.sns.Load()) |
| + } |
| + |
| + // No sinks: nil, and one line in the journal however often it is asked. |
| + var buf bytes.Buffer |
| + prev := log.Writer() |
| + log.SetOutput(&buf) |
| + t.Cleanup(func() { log.SetOutput(prev) }) |
| + armPaging(nil) |
| + var none *alertSinks |
| + for range 2 { |
| + if err := PageOperator(ctx, alerting.LevelIncident, "s", "b"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := none.PageOperator(ctx, alerting.LevelIncident, "s", "b"); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if n := strings.Count(buf.String(), "no sink configured"); n != 1 { |
| + t.Fatalf("no-sink notice logged %d times, want once:\n%s", n, buf.String()) |
| + } |
| + if len(pages) != 3 { |
| + t.Fatalf("an unarmed page reached the old sinks: %+v", pages) |
| + } |
| +} |
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go| index 9e92c102d..254b74868 100644 |
| --- a/internal/host/metrics_loop.go |
| +++ b/internal/host/metrics_loop.go |
| @@ -9,17 +9,15 @@ |
| package host |
| import ( |
| - "bytes" |
| "context" |
| "encoding/json" |
| "errors" |
| - "fmt" |
| "log" |
| - "net/http" |
| "os" |
| "path/filepath" |
| "time" |
| + "github.com/carlosframework/platform/internal/alerting" |
| "github.com/carlosframework/platform/internal/metrics" |
| "github.com/carlosframework/platform/internal/release" |
| "github.com/carlosframework/platform/internal/status" |
| @@ -592,18 +590,7 @@ func currentAlertEtag(ctx context.Context, s release.Store, key string) (string, |
| // httpPostWebhook is the production webhook post func: a 5s-timeout POST |
| // with a JSON content type. Any non-2xx status is an error — the caller |
| // (alertTick) only logs it, but it still needs to know delivery failed. |
| -func httpPostWebhook(url string, body []byte) error { |
| - client := &http.Client{Timeout: 5 * time.Second} |
| - resp, err := client.Post(url, "application/json", bytes.NewReader(body)) |
| - if err != nil { |
| - return err |
| - } |
| - defer resp.Body.Close() |
| - if resp.StatusCode < 200 || resp.StatusCode >= 300 { |
| - return fmt.Errorf("host: webhook: non-2xx status %s", resp.Status) |
| - } |
| - return nil |
| -} |
| +func httpPostWebhook(url string, body []byte) error { return alerting.PostJSON(url, body) } |
| // realSampleProbes wires sampleProbes to the actual OS/cgroup readers in |
| // metrics_probes.go and probes.go — the production seam metricsLoop uses, |
| @@ -803,6 +790,8 @@ func metricsLoop(execRoot string, stampedUnits func() []string) { |
| rules := metrics.RulesFromEnv(os.Getenv) |
| logRejectedAlertOverrides(os.Getenv) |
| sinks := buildAlertSinks(ctx, os.Getenv, &processAlertCounters, newSNSClient) |
| + // The moves code's operator pages go out on the same sinks. |
| + armPaging(sinks) |
| log.Printf("agent: metrics loop armed (60s ticks, 5min flush), publishing to %s", target) |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index 5285cc85c..f1ebed17b 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -232,6 +232,7 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, |
| ReadyDelta: readyDelta, |
| Enabled: moves.Enabled(os.Getenv), |
| claimRecords: writes.Enabled, |
| + Page: PageOperator, |
| } |
| w := &movesWorker{mirrors: map[string]lease.MirrorStore{}} |
| if cfg.Enabled { |
| @@ -584,6 +585,13 @@ type moveCfg struct { |
| LogEvent func(ctx context.Context, rec provision.Record, event string) |
| rbFault func(point string) error |
| + // Page pages an operator (moves spec §13, K11): PageOperator in |
| + // production (armMovesPass), nil in tests, where nothing pages. The box |
| + // arms call it where a MoveMarkStuck lands (pageStuck): the rollback's |
| + // return on the target, and the recovery, its dirty return's install |
| + // and the revert on the source. |
| + Page func(ctx context.Context, level, subject, body string) error |
| + |
| // The source's recovery after a rollback (§7.8, movesrecover.go). |
| // ReleaseHeld opens the row: registry.ReleaseHold under the op that |
| // holds it, touching the stage-requested marker when it landed |
diff --git a/internal/host/movesrecover.go b/internal/host/movesrecover.go| index d51924ce3..67793c013 100644 |
| --- a/internal/host/movesrecover.go |
| +++ b/internal/host/movesrecover.go |
| @@ -303,6 +303,7 @@ func (r *recoverRun) stuck(ctx context.Context, reason string, why error) (bool, |
| return false, fmt.Errorf("mark the recovery stuck: %w", err) |
| } |
| r.w.logEvent(ctx, r.rec, "source-stuck") |
| + r.w.pageStuck(ctx, r.rec, "source", reason) |
| nudgeSerial(ctx, r.s) |
| return true, nil |
| } |
| @@ -579,6 +580,7 @@ func (r *recoverRun) install(ctx context.Context) (bool, error) { |
| if refused != "" { |
| log.Printf("moves: %s: the return of %s stopped for an operator: %s", host, r.rb, refused) |
| r.w.logEvent(ctx, r.rec, "source-stuck") |
| + r.w.pageStuck(ctx, r.rec, "source", refused) |
| } |
| return advanced, err |
| } |
diff --git a/internal/host/movesrecover_test.go b/internal/host/movesrecover_test.go| index bd270da0b..928368cb6 100644 |
| --- a/internal/host/movesrecover_test.go |
| +++ b/internal/host/movesrecover_test.go |
| @@ -151,6 +151,7 @@ func (h *recoverHarness) sourceCfg(t *testing.T) moveCfg { |
| return true, nil |
| } |
| c.LogEvent = func(_ context.Context, _ provision.Record, event string) { h.sEvents = append(h.sEvents, event) } |
| + c.Page = h.pages.hook() |
| c.recoveryProbe = func(ctx context.Context, _ release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) { |
| h.probes = append(h.probes, req) |
| if h.probe != nil { |
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go| index d57869f32..3a2f6e0c4 100644 |
| --- a/internal/host/movesrevert.go |
| +++ b/internal/host/movesrevert.go |
| @@ -738,6 +738,7 @@ func (r *revertRun) stuck(ctx context.Context, reason string, why error) (bool, |
| return false, err |
| } |
| r.w.logEvent(ctx, r.rec, "source-stuck") |
| + r.w.pageStuck(ctx, r.rec, "source", reason) |
| nudgeSerial(ctx, r.s) |
| return true, nil |
| } |
diff --git a/internal/host/movesrollback.go b/internal/host/movesrollback.go| index c9d5d85d0..91413190a 100644 |
| --- a/internal/host/movesrollback.go |
| +++ b/internal/host/movesrollback.go |
| @@ -14,6 +14,7 @@ import ( |
| "strings" |
| "time" |
| + "github.com/carlosframework/platform/internal/alerting" |
| "github.com/carlosframework/platform/internal/appconfig" |
| "github.com/carlosframework/platform/internal/hostclaim" |
| "github.com/carlosframework/platform/internal/hostguard" |
| @@ -92,8 +93,9 @@ import ( |
| // (ParkLeases), which production does not wire, so rollbackTargetArmed |
| // answers false there and the pass logs a standing notice. The decision is |
| // not gated (J10). The move's event log (moves.LogEvent, Plan 6) is the |
| -// LogEvent hook: nil in production, where the event is a log line. Nothing |
| -// here pages: the release deadline is the console's. |
| +// LogEvent hook: nil in production, where the event is a log line. The |
| +// stuck mark pages an operator (pageStuck, the Page hook); the release |
| +// deadline is the console's. |
| // rollbackStepsPerPass bounds the phases one pass runs. |
| const rollbackStepsPerPass = 8 |
| @@ -135,6 +137,37 @@ func (w *movesWorker) logEvent(ctx context.Context, rec provision.Record, event |
| log.Printf("moves: %s: %s %s (the move log is Plan 6's)", rec.Host, id, event) |
| } |
| +// pageStuck pages an operator once a MoveMarkStuck has landed (moves spec |
| +// §13: stuck is an incident; K11). role is this box's part in the move, |
| +// "source" or "target", and reason the recorded stuck sentence. The page |
| +// names the host, the move, the box and the reason, and nothing else: no |
| +// path, no error text, no secret. A replay that lands the same mark again |
| +// pages again. Unwired (nil Page), nothing pages; a failed page is logged |
| +// and never stops the arm. |
| +func (w *movesWorker) pageStuck(ctx context.Context, rec provision.Record, role, reason string) { |
| + f := w.cfg.Page |
| + if f == nil { |
| + return |
| + } |
| + id := "" |
| + if rec.Move != nil { |
| + id = rec.Move.ID |
| + } |
| + subject, body := stuckPage(rec.Host, id, w.cfg.Box, role, reason) |
| + if err := f(ctx, alerting.LevelIncident, subject, body); err != nil { |
| + log.Printf("moves: %s: the page for the stuck mark of %s: %v", rec.Host, id, err) |
| + } |
| +} |
| + |
| +// stuckPage is the stuck mark's page: the subject (MV-N313) and the body |
| +// (MV-N314). |
| +func stuckPage(host, id, box, role, reason string) (subject, body string) { |
| + subject = fmt.Sprintf("carlos move stuck: %s", host) |
| + body = fmt.Sprintf("The move %s of %s stopped for an operator on %s, its %s: %s. Run carlos move resolve --host %s --retry once the cause is fixed.", |
| + id, host, box, role, reason, host) |
| + return subject, body |
| +} |
| + |
| // rollBackLanding is the rollback decision on the target (§7.8; J10, J11): |
| // a landing marked for a rollback, or past landingDeadlineOf at any phase, |
| // is rolled back by MoveRollBack, and the claim follows. done is true when |
| @@ -862,6 +895,7 @@ func (r *rollbackRun) stuck(ctx context.Context, refused *returnRefusal) (bool, |
| return false, fmt.Errorf("mark the return stuck: %w", err) |
| } |
| r.w.logEvent(ctx, r.rec, "target-stuck") |
| + r.w.pageStuck(ctx, r.rec, "target", refused.reason) |
| nudgeSerial(ctx, r.s) |
| return true, nil |
| } |
diff --git a/internal/host/movesrollback_test.go b/internal/host/movesrollback_test.go| index 022a5461f..43581e789 100644 |
| --- a/internal/host/movesrollback_test.go |
| +++ b/internal/host/movesrollback_test.go |
| @@ -42,6 +42,7 @@ type rollbackHarness struct { |
| tchecks int // the target's final checkpoints |
| tpos uint64 // the target's local position |
| events []string |
| + pages pagesSeen // the operator pages (moveCfg.Page), the target's and the source's |
| rbAt string |
| rbDone bool |
| rbHits []string |
| @@ -135,6 +136,7 @@ func (h *rollbackHarness) rollbackCfg(t *testing.T) moveCfg { |
| return true, nil |
| } |
| c.LogEvent = func(_ context.Context, _ provision.Record, event string) { h.events = append(h.events, event) } |
| + c.Page = h.pages.hook() |
| c.rbFault = func(point string) error { |
| h.rbHits = append(h.rbHits, point) |
| if h.rbHook != nil { |
diff --git a/internal/host/movesstuckpage_test.go b/internal/host/movesstuckpage_test.go| new file mode 100644 |
| index 000000000..786283953 |
| --- /dev/null |
| +++ b/internal/host/movesstuckpage_test.go |
| @@ -0,0 +1,115 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// pagesSeen records the pages a box arm sends through moveCfg.Page. |
| +type pagesSeen []struct{ level, subject, body string } |
| + |
| +func (p *pagesSeen) hook() func(context.Context, string, string, string) error { |
| + return func(_ context.Context, level, subject, body string) error { |
| + *p = append(*p, struct{ level, subject, body string }{level, subject, body}) |
| + return nil |
| + } |
| +} |
| + |
| +// stuckPageBody is the page the record's stuck mark sends (MV-N314), spelled |
| +// out here rather than built by the code under test. |
| +func stuckPageBody(box, role, reason string) string { |
| + return "The move " + p1ID + " of " + p1Host + " stopped for an operator on " + box + ", its " + role + ": " + reason + |
| + ". Run carlos move resolve --host " + p1Host + " --retry once the cause is fixed." |
| +} |
| + |
| +// pagedOnce asserts exactly one incident page for the stuck mark, carrying |
| +// no path and no secret. |
| +func pagedOnce(t *testing.T, pages pagesSeen, box, role, reason string) { |
| + t.Helper() |
| + if len(pages) != 1 { |
| + t.Fatalf("pages %+v, want one", pages) |
| + } |
| + p := pages[0] |
| + if p.level != "incident" || p.subject != "carlos move stuck: "+p1Host || p.body != stuckPageBody(box, role, reason) { |
| + t.Fatalf("page %+v\nwant body %q", p, stuckPageBody(box, role, reason)) |
| + } |
| + if strings.Contains(p.subject+p.body, "/") { |
| + t.Fatalf("a page carries a path: %+v", p) |
| + } |
| +} |
| + |
| +// TestStuckMarksPageAnOperator (moves spec §13, K11): wherever a box arm's |
| +// MoveMarkStuck lands, the operator is paged once, as an incident, with the |
| +// recorded reason; a pass over the stuck record pages nothing more. The |
| +// hook is nil in the other tests and in production until armMovesPass sets |
| +// it to PageOperator. |
| +func TestStuckMarksPageAnOperator(t *testing.T) { |
| + t.Run("the target's return", func(t *testing.T) { |
| + h := newRollbackHarness(t) |
| + db := provision.DefaultDBPath(h.tdir, p1Host) |
| + if err := os.Rename(db, db+".aside"); err != nil { |
| + t.Fatal(err) |
| + } |
| + h.rollBack(t) |
| + h.rollbackPass(t) |
| + h.stuckAtReturning(t, provision.StuckReturnTargetDatabase) |
| + h.rollbackPass(t) |
| + pagedOnce(t, h.pages, "uk-box", "target", provision.StuckReturnTargetDatabase) |
| + }) |
| + t.Run("the source's recovery", func(t *testing.T) { |
| + h := newRecoverHarness(t, false) |
| + h.released(t) |
| + if err := os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("changed on the source"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + h.pages = nil |
| + h.srcPass(t) |
| + if m := h.read(t, p1Host).Move; m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch { |
| + t.Fatalf("not stuck: %+v", m) |
| + } |
| + h.srcPass(t) |
| + pagedOnce(t, h.pages, "ie-box", "source", provision.StuckBaselineMismatch) |
| + }) |
| + t.Run("the source's dirty return install", func(t *testing.T) { |
| + h := newRecoverHarness(t, true) |
| + h.released(t) |
| + sentinel := filepath.Join(h.sdir, p1Host, "sentinel.txt") |
| + h.landFault = func(p string) error { |
| + if p == "after-"+landStepRetired { |
| + if err := os.MkdirAll(filepath.Dir(sentinel), 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + return os.WriteFile(sentinel, []byte("not the return's"), 0o644) |
| + } |
| + return nil |
| + } |
| + h.pages = nil |
| + h.srcPass(t) |
| + if m := h.read(t, p1Host).Move; m.State != provision.MoveStuck || m.StuckReason != provision.StuckReturnFilesOccupied { |
| + t.Fatalf("not stuck: %+v", m) |
| + } |
| + h.srcPass(t) |
| + pagedOnce(t, h.pages, "ie-box", "source", provision.StuckReturnFilesOccupied) |
| + }) |
| + t.Run("the source's revert", func(t *testing.T) { |
| + h := newRevertHarness(t) |
| + h.fromStopped(t) |
| + h.arm(t) |
| + var pages pagesSeen |
| + h.cfg.Page = pages.hook() |
| + if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("changed while stopped"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + h.pass(t) |
| + if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch { |
| + t.Fatalf("not stuck: %+v", m) |
| + } |
| + h.pass(t) |
| + pagedOnce(t, pages, "ie-box", "source", provision.StuckBaselineMismatch) |
| + }) |
| +} |
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go| index a147485c9..f62eb3dd5 100644 |
| --- a/internal/provision/plan9_b8_census_test.go |
| +++ b/internal/provision/plan9_b8_census_test.go |
| @@ -508,7 +508,7 @@ var plan9B8CensusInventory = []censusEntry{ |
| // Records probe inputs through SetRecordMove; phase and stuck helpers use the same writer. |
| {"internal/host/movesrecover.go", "recoverRun.probing", "record-writer", "key:SetRecordMove", "7184b393fc356deb85181e25e4cafd61079598bf7ea4dafc28f90cbe26d50a31"}, |
| // Marks recovery stuck through SetRecordMove; event and serial writes are not record objects. |
| - {"internal/host/movesrecover.go", "recoverRun.stuck", "record-writer", "key:SetRecordMove", "e2942947ddb1559e44b494e9900fa82ebbe3782a86333167797f4a94375a4a6c"}, |
| + {"internal/host/movesrecover.go", "recoverRun.stuck", "record-writer", "key:SetRecordMove", "4ea620140bbc75b127a893926eefcf58685b40c8362f86ba94f22931a05b449a"}, |
| // Records revert inputs through SetRecordMove; stuck uses the reviewed record helper and journals are local. |
| {"internal/host/movesrevert.go", "revertRun.probing", "record-writer", "key:SetRecordMove", "67f15935f08fe9111b713641724093661e8c9dd2ee421287a1fc6cbe0509faf4"}, |
| // Applies revert record mutations only through SetRecordMove and its sequenced writer. |
| @@ -520,7 +520,7 @@ var plan9B8CensusInventory = []censusEntry{ |
| // Advances the target rollback phase through SetRecordMove and its sequenced writer. |
| {"internal/host/movesrollback.go", "rollbackRun.phase", "record-writer", "key:SetRecordMove", "76f1efb1076029e5da4c0cc0d750fbac5e1c94a6014c2b1e4af241797350827b"}, |
| // Marks the return stuck through SetRecordMove; event and serial writes are not record objects. |
| - {"internal/host/movesrollback.go", "rollbackRun.stuck", "record-writer", "key:SetRecordMove", "8421f9feb384491bdb9d9cf3f722887b522bb70c41892526f7578e85654b0541"}, |
| + {"internal/host/movesrollback.go", "rollbackRun.stuck", "record-writer", "key:SetRecordMove", "84fcea399a55d77ceee17e35652ef269e5811453ffff62dff43208b68538b1ed"}, |
| // Validates the pointer-digest marker key and Put/Delete method names without writing. |
| {"internal/hostops/mutation_classifier.go", "classifyPointerDigestMarker", "reference", "str:Put", "08111b5db5d98e788d1eef5861ad9c861cf926c5e1d5d54717fd650d2dcf9fa8"}, |
| // Renders systemd paths under /data/instances and /run/carlos/instances; performs no store write. |