carlos / platform Public

self-serve pages: an attempt with no sink armed is not a page sent (final review F13)

alerting.Armed reports whether the armed sinks can deliver. selfServePage
records nothing and sends nothing while no sink is armed, so NewServer's
bootstrap orphan scan (which runs before cmd/console arms the sinks) no
longer charges the ten-minute throttle; the pass's first tick, after the
arm, pages and records.

Witness: TestOrphanReservationIsPagedAtBootstrap, rewritten in production
start order with a counting sink armed after NewServer. The e2e world arms
a counting sink as cmd/console does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 1 day ago 5591ccb256ff61cc684a82eb848f2569cf3872cc parent ec91cf9

4 files changed, +79 −3

internal/alerting/alerting.go +11 −0
internal/console/selfserve_pages.go +9 −1
internal/e2e/selfserve/cut_test.go +16 −2
internal/e2e/selfserve/job_world_test.go +43 −0

diff --git a/internal/alerting/alerting.go b/internal/alerting/alerting.go

index ec088e105..3270a99d9 100644

--- a/internal/alerting/alerting.go

+++ b/internal/alerting/alerting.go

@@ -179,6 +179,17 @@ func Arm(s *Sinks) {

noSinkLogged.Store(false)

}

+// Armed reports whether the armed sinks can deliver a page (a webhook or

+// SNS is on). A caller that throttles its pages checks it first, so an

+// attempt made before the process armed its sinks, or with none

+// configured, is not counted as a page sent.

+func Armed() bool {

+ armedMu.RLock()

+ s := armed

+ armedMu.RUnlock()

+ return s.HasWebhook() || s.HasSNS()

// Page pages an operator on the armed sinks (Sinks.Page).

func Page(ctx context.Context, level, subject, body string) error {

armedMu.RLock()

diff --git a/internal/console/selfserve_pages.go b/internal/console/selfserve_pages.go

index 5129f97ae..9ead9433a 100644

--- a/internal/console/selfserve_pages.go

+++ b/internal/console/selfserve_pages.go

@@ -6,7 +6,8 @@ package console

// one rule per move and condition, selfserve:<id>:<condition>, at most once

// every 10 minutes. The attempt is recorded before the page is sent, as the

// move pass's pages are: a crash between the two loses one page, never

-// sends two.

+// sends two. An attempt while no sink is armed (alerting.Armed) is not

+// recorded and sends nothing.

import (

"context"

@@ -50,6 +51,13 @@ func (s *Server) selfServePage(ctx context.Context, id, condition, subject, body

return

}

rule := selfServePageRule(id, condition)

+ if !alerting.Armed() {

+ // No sink can deliver it (the process has not armed its sinks yet,

+ // or none is configured): not an attempt, so the throttle does not

+ // hold back the first page a sink could deliver (final review F13).

+ log.Printf("console: self-serve move %s: page %s not sent: no alert sink is armed", id, rule)

+ return

+ }

last, paged, err := s.store.MovePageAttempt(rule)

if err != nil {

log.Printf("console: self-serve move %s: the page record for %s: %v", id, rule, err)

diff --git a/internal/e2e/selfserve/cut_test.go b/internal/e2e/selfserve/cut_test.go

index 5ab625eda..995a1f669 100644

--- a/internal/e2e/selfserve/cut_test.go

+++ b/internal/e2e/selfserve/cut_test.go

@@ -8,6 +8,7 @@ import (

"testing"

"time"

+ "github.com/carlosframework/platform/internal/alerting"

"github.com/carlosframework/platform/internal/console"

"github.com/carlosframework/platform/internal/host"

"github.com/carlosframework/platform/internal/moves"

@@ -787,15 +788,28 @@ func TestSwitchOffMidFlightCompletesAndRendersAfterRestart(t *testing.T) {

// TestOrphanReservationIsPagedAtBootstrap: a reservation no move owns, past

// the grace period, pages from the console's start and every tick after,

-// throttled.

+// throttled. In production's start order (final review F13): NewServer

+// (its recovery scan), then cmd/console arms the alert sinks, then the pass

+// ticks; the bootstrap page reaches the sink armed after NewServer, and

+// only then does the throttle hold later attempts back.

func TestOrphanReservationIsPagedAtBootstrap(t *testing.T) {

w := newJobWorld(t)

orphan := "mv_orphanorphanorphanorphanor"

w.putReservation(moves.OfflineReservation{Kind: moves.OfflineKindMove, ID: orphan, Host: host.E2EHost, Account: w.acct, App: host.E2EApp,

ReservedAt: time.Now().Add(-time.Hour).UTC()})

+ alerting.Arm(nil) // a process before cmd/console arms its sinks

w.restart()

+ sink := w.armSink()

+ w.tick()

+ if n := sink.count(ssPageSubjectFor(host.E2EHost)); n != 1 {

+ t.Fatalf("the bootstrap page reached the armed sink %d times, want once", n)

+ }

if _, ok := w.paged(orphan, "reservation"); !ok {

- t.Fatal("no page at the console's start")

+ t.Fatal("the delivered page was not recorded for the throttle")

+ }

+ w.tick()

+ if n := sink.count(ssPageSubjectFor(host.E2EHost)); n != 1 {

+ t.Fatalf("the throttle let %d pages through inside its 10 minutes", n)

}

diff --git a/internal/e2e/selfserve/job_world_test.go b/internal/e2e/selfserve/job_world_test.go

index 0b5ebaa6e..653f2ff56 100644

--- a/internal/e2e/selfserve/job_world_test.go

+++ b/internal/e2e/selfserve/job_world_test.go

@@ -14,6 +14,7 @@ import (

"testing"

"time"

+ "github.com/carlosframework/platform/internal/alerting"

"github.com/carlosframework/platform/internal/console"

"github.com/carlosframework/platform/internal/host"

"github.com/carlosframework/platform/internal/moves"

@@ -229,6 +230,44 @@ type jobWorld struct {

memberToken, memberFront string

}

+// pageSink counts the pages delivered to the world's webhook sink, by

+// subject.

+type pageSink struct {

+ mu sync.Mutex

+ subjects []string

+func (p *pageSink) post(_ string, body []byte) error {

+ var m struct{ Subject string }

+ _ = json.Unmarshal(body, &m)

+ p.mu.Lock()

+ p.subjects = append(p.subjects, m.Subject)

+ p.mu.Unlock()

+ return nil

+func (p *pageSink) count(subject string) int {

+ p.mu.Lock()

+ defer p.mu.Unlock()

+ n := 0

+ for _, s := range p.subjects {

+ if s == subject {

+ n++

+ }

+ return n

+// armSink arms a counting webhook sink, as cmd/console arms its sinks.

+func (w *jobWorld) armSink() *pageSink {

+ p := &pageSink{}

+ alerting.Arm(&alerting.Sinks{WebhookURL: "http://127.0.0.1/alerts", Post: p.post})

+ return p

+// ssPageSubjectFor is the self-serve pages' subject for host.

+func ssPageSubjectFor(h string) string { return "carlos self-serve move needs an operator: " + h }

// switchOf is the console's CARLOS_SELF_SERVE_MOVES.

func switchOf(off bool) string {

if off {

@@ -270,6 +309,10 @@ func newJobWorld(t *testing.T) *jobWorld {

}

})

console.TestHooks.Fault(w.fault)

+ // The alert sinks, armed as cmd/console arms them after NewServer: a

+ // page is recorded for its throttle only when a sink can deliver it.

+ w.armSink()

+ t.Cleanup(func() { alerting.Arm(nil) })

// The boxes publish their edges files and status blocks.

w.boxes()

return w