amadan

carlos / platform Public

feat/claim-loss-pages

Squashed from feat/claim-loss-pages into main.
Co-Authored-By: claude <claude~paul@keymail.dev>
codex pushed by claude 39986908150d2e3c90477de4320c0d2941ffaac6 parent 8ebd9d8
13 files changed, +959 −9
  • approved_moves_copy_test.go +3 −0
  • cmd/console/main.go +3 −4
  • cmd/console/movesswitch_test.go +3 −3
  • internal/console/claim_authority_census_test.go +5 −1
  • internal/console/claim_loss_pages.go +166 −0
  • internal/console/claim_loss_pages_test.go +730 −0
  • internal/console/claim_runtime.go +1 −0
  • internal/console/server.go +2 −0
  • internal/console/store_gate.go +19 −0
  • internal/console/testdata/claim-authority-sites.txt +10 −0
  • internal/hostops/recovery.go +5 −0
  • internal/provision/plan9_b8_census_test.go +2 −0
  • testdata/claim-moves-copy.md +10 −1

CI passed — run details

CI log 
=== gate ===
cd launchctx && go vet ./... && go test ./... -count=1
ok  	amadan.net/carlos/platform.git/launchctx	0.011s
?   	amadan.net/carlos/platform.git/launchctx/controlname	[no test files]
?   	amadan.net/carlos/platform.git/launchctx/wireid	[no test files]
go vet ./...
gofmt -l .: clean
go test ./... -count=1 -timeout 30m
ok  	github.com/carlosframework/platform	317.550s
?   	github.com/carlosframework/platform/cmd/changelog	[no test files]
ok  	github.com/carlosframework/platform/cmd/console	3.243s
ok  	github.com/carlosframework/platform/cmd/moveprobe	93.285s
ok  	github.com/carlosframework/platform/internal/activator	209.026s
ok  	github.com/carlosframework/platform/internal/agentsock	0.033s
ok  	github.com/carlosframework/platform/internal/alerting	0.006s
ok  	github.com/carlosframework/platform/internal/analytics	0.022s
ok  	github.com/carlosframework/platform/internal/appconfig	2.246s
ok  	github.com/carlosframework/platform/internal/apperr	0.124s
ok  	github.com/carlosframework/platform/internal/applog	0.206s
ok  	github.com/carlosframework/platform/internal/awsinv	0.019s
ok  	github.com/carlosframework/platform/internal/boxchoice	0.010s
ok  	github.com/carlosframework/platform/internal/boxid	0.692s
?   	github.com/carlosframework/platform/internal/boxid/boxidtest	[no test files]
?   	github.com/carlosframework/platform/internal/boxid/internal/signtest	[no test files]
ok  	github.com/carlosframework/platform/internal/boxvolumes	0.012s
ok  	github.com/carlosframework/platform/internal/caps	0.016s
ok  	github.com/carlosframework/platform/internal/certwarm	0.003s
ok  	github.com/carlosframework/platform/internal/cgroup	0.203s
ok  	github.com/carlosframework/platform/internal/changelog	0.265s
ok  	github.com/carlosframework/platform/internal/clamd	0.061s
?   	github.com/carlosframework/platform/internal/cliflag	[no test files]
ok  	github.com/carlosframework/platform/internal/console	355.830s
?   	github.com/carlosframework/platform/internal/controlname	[no test files]
?   	github.com/carlosframework/platform/internal/controlsock	[no test files]
ok  	github.com/carlosframework/platform/internal/dbdoor	6.512s
ok  	github.com/carlosframework/platform/internal/dbdoor/dbdoortest	0.011s
ok  	github.com/carlosframework/platform/internal/dbverify	0.006s
ok  	github.com/carlosframework/platform/internal/dnsrec	0.006s
ok  	github.com/carlosframework/platform/internal/docsite	0.006s
ok  	github.com/carlosframework/platform/internal/domains	1.980s
ok  	github.com/carlosframework/platform/internal/doorkey	0.007s
ok  	github.com/carlosframework/platform/internal/economics	0.006s
ok  	github.com/carlosframework/platform/internal/edge	25.698s
?   	github.com/carlosframework/platform/internal/edgeproof	[no test files]
ok  	github.com/carlosframework/platform/internal/edgesfile	0.007s
ok  	github.com/carlosframework/platform/internal/fleetapi	0.125s
ok  	github.com/carlosframework/platform/internal/gate	1.655s
ok  	github.com/carlosframework/platform/internal/genselect	0.016s
ok  	github.com/carlosframework/platform/internal/geo	0.012s
ok  	github.com/carlosframework/platform/internal/geoip	0.037s
ok  	github.com/carlosframework/platform/internal/host	54.422s
ok  	github.com/carlosframework/platform/internal/hostactors	0.420s
ok  	github.com/carlosframework/platform/internal/hostbox	1.400s
ok  	github.com/carlosframework/platform/internal/hostclaim	0.235s
ok  	github.com/carlosframework/platform/internal/hostevidence	0.005s
ok  	github.com/carlosframework/platform/internal/hostexec	0.311s
ok  	github.com/carlosframework/platform/internal/hostfence	0.044s
ok  	github.com/carlosframework/platform/internal/hostfiles	4.155s
ok  	github.com/carlosframework/platform/internal/hostfloor	0.018s
ok  	github.com/carlosframework/platform/internal/hostguard	0.146s
ok  	github.com/carlosframework/platform/internal/hostops	1.346s
?   	github.com/carlosframework/platform/internal/hostops/fencerelease	[no test files]
ok  	github.com/carlosframework/platform/internal/hostops/internal/fencerelease	0.002s
ok  	github.com/carlosframework/platform/internal/hostquarantine	0.004s
ok  	github.com/carlosframework/platform/internal/hostresolve	0.141s
ok  	github.com/carlosframework/platform/internal/hostwrite	0.004s
ok  	github.com/carlosframework/platform/internal/infraview	0.002s
ok  	github.com/carlosframework/platform/internal/instancectl	1.704s
ok  	github.com/carlosframework/platform/internal/instanceexec	2.682s
ok  	github.com/carlosframework/platform/internal/instancefs	0.003s
ok  	github.com/carlosframework/platform/internal/instancekey	0.003s
ok  	github.com/carlosframework/platform/internal/instanceupdates	0.006s
ok  	github.com/carlosframework/platform/internal/layout	1.328s
ok  	github.com/carlosframework/platform/internal/lease	9.431s
ok  	github.com/carlosframework/platform/internal/lease/leasetest	0.008s
ok  	github.com/carlosframework/platform/internal/ledger	0.015s
ok  	github.com/carlosframework/platform/internal/mail	0.161s
ok  	github.com/carlosframework/platform/internal/maintenance	0.076s
ok  	github.com/carlosframework/platform/internal/maintenanceexec	5.532s
ok  	github.com/carlosframework/platform/internal/maintenanceguard	0.072s
ok  	github.com/carlosframework/platform/internal/metrics	0.007s
ok  	github.com/carlosframework/platform/internal/moves	21.077s
ok  	github.com/carlosframework/platform/internal/moves/archive	0.729s
ok  	github.com/carlosframework/platform/internal/objstore	0.156s
ok  	github.com/carlosframework/platform/internal/ops	0.151s
ok  	github.com/carlosframework/platform/internal/park	0.182s
ok  	github.com/carlosframework/platform/internal/privatefile	0.007s
ok  	github.com/carlosframework/platform/internal/provision	3.140s
ok  	github.com/carlosframework/platform/internal/publicrelease	0.009s
?   	github.com/carlosframework/platform/internal/regionname	[no test files]
ok  	github.com/carlosframework/platform/internal/registry	3.680s
ok  	github.com/carlosframework/platform/internal/relay	7.222s
ok  	github.com/carlosframework/platform/internal/release	18.439s
ok  	github.com/carlosframework/platform/internal/rollout	3.232s
ok  	github.com/carlosframework/platform/internal/rolloutwords	0.003s
ok  	github.com/carlosframework/platform/internal/rootaws	0.006s
ok  	github.com/carlosframework/platform/internal/s3compat	0.005s
ok  	github.com/carlosframework/platform/internal/safefile	0.004s
ok  	github.com/carlosframework/platform/internal/schedule	2.198s
ok  	github.com/carlosframework/platform/internal/sim/hostlife	2.350s
ok  	github.com/carlosframework/platform/internal/sim/litestreamfacts	3.432s
ok  	github.com/carlosframework/platform/internal/status	0.039s
ok  	github.com/carlosframework/platform/internal/steering	0.005s
ok  	github.com/carlosframework/platform/internal/stopevidence	0.088s
ok  	github.com/carlosframework/platform/internal/svcrole	0.002s
ok  	github.com/carlosframework/platform/internal/telemetry	0.004s
ok  	github.com/carlosframework/platform/internal/termfence	0.004s
ok  	github.com/carlosframework/platform/internal/tunnelclient	0.004s
ok  	github.com/carlosframework/platform/internal/unixdial	0.003s
ok  	github.com/carlosframework/platform/internal/updatewords	0.002s
ok  	github.com/carlosframework/platform/internal/usage	0.004s
ok  	github.com/carlosframework/platform/internal/version	0.002s
ok  	github.com/carlosframework/platform/internal/vet	1.542s
=== cross ===
--- shard 0 of 5 ---
cross-compiling darwin/arm64
--- shard 1 of 5 ---
cross-compiling darwin/amd64
--- shard 2 of 5 ---
cross-compiling linux/arm64
--- shard 3 of 5 ---
cross-compiling windows/amd64
--- shard 4 of 5 ---
cross-compiling windows/arm64
=== minio-contract ===
minio contract: arm64, minio RELEASE.2025-09-07T16-13-09Z, mc RELEASE.2025-08-13T08-35-41Z
  minio: cached, digest verified
  mc: cached, digest verified
  minio healthy on 127.0.0.1:20999
Added `local` successfully.
Bucket created successfully `local/contract-bucket`.
Bucket created successfully `local/denied-bucket`.
Created policy `contract-only` successfully.
Added user `scopeduser` successfully.
Attached Policies: [contract-only]
To User: scopeduser
  buckets and scoped user ready
=== RUN   TestDeliveredContractMinio
=== RUN   TestDeliveredContractMinio/put
=== RUN   TestDeliveredContractMinio/get_round-trips_the_body
=== RUN   TestDeliveredContractMinio/list_shows_the_key
=== RUN   TestDeliveredContractMinio/head_and_list_agree_on_the_version
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed
=== RUN   TestDeliveredContractMinio/delete_removes_it
=== RUN   TestDeliveredContractMinio/cross-bucket_denial
--- PASS: TestDeliveredContractMinio (0.02s)
    --- PASS: TestDeliveredContractMinio/put (0.00s)
    --- PASS: TestDeliveredContractMinio/get_round-trips_the_body (0.00s)
    --- PASS: TestDeliveredContractMinio/list_shows_the_key (0.00s)
    --- PASS: TestDeliveredContractMinio/head_and_list_agree_on_the_version (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route (0.01s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed (0.00s)
    --- PASS: TestDeliveredContractMinio/delete_removes_it (0.00s)
    --- PASS: TestDeliveredContractMinio/cross-bucket_denial (0.00s)
PASS
ok  	github.com/carlosframework/platform/internal/objstore	0.022s
=== RUN   TestT4StorageProbeMinio
2026/10/12 16:51:30 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DDD655C21FD559, HostID: 7c420053e0fbcd7f41653fdd92ffa2802c903d05303b1aa15adc2e8bb65d183f, api error SignatureDoesNotMatch: The request signature we calculated does not match the signature you provided. Check your key and signing method.
--- PASS: TestT4StorageProbeMinio (0.01s)
PASS
ok  	github.com/carlosframework/platform/internal/host	0.020s
minio contract: PASS
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index eef047380..e15cbdf49 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -41,6 +41,7 @@ const movesCopyRecordPath = "docs/superpowers/plans/2026-10-06-moves-phase1-copy
var movesCopyIDs = []string{
"MV-N391",
"HC5-N1", "HC5-N2", "HC5-N3",
+ "HC5-N4", "HC5-N5", "HC5-N6", "HC5-N7",
"M-C1", "M-C2", "M-C3", "M-C4", "M-C5", "M-C6", "M-C7", "M-C8", "M-C9",
"M-C10", "M-C11", "M-C12", "M-C13", "M-C14", "M-C15", "M-C16", "M-C17",
"M-C18", "M-C19", "M-C20", "M-C21", "M-C22", "MV-N1", "MV-N2", "MV-N3",
@@ -1152,6 +1153,8 @@ func TestMovesTextAppliedPlaceholders(t *testing.T) {
// copy" task appends the files it touched; TestMovesScanFilesCoverTheDiff
// fails when the list falls behind.
var movesScanFiles = []string{
+ // Claim-owned pages share the alert transport; keep their drafts in the copy census.
+ "internal/console/claim_loss_pages.go",
"internal/host/claimhandover_crossbox.go",
"internal/host/claimhandover_recovery.go",
"internal/host/claimretire_local.go",
diff --git a/cmd/console/main.go b/cmd/console/main.go
index f4b243819..07d52df85 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -1350,11 +1350,10 @@ func applyMovesSwitch(cfg *console.Config, getenv func(string) string) {
}
// armMovesPaging arms alerting's sinks from CARLOS_ALERT_WEBHOOK and
-// CARLOS_ALERT_SNS_TOPIC_ARN, the agent's variables, for the console's move
-// pass (moves Plan 5 Task 7, delta M14). Only a host-claim console with the
-// moves switch on runs that pass, so only it arms; it reports whether it did.
+// CARLOS_ALERT_SNS_TOPIC_ARN for claim recovery and its optional move pass.
+// Claim loss pages run even when moves are disabled.
func armMovesPaging(ctx context.Context, cfg console.Config, getenv func(string) string) bool {
- if !cfg.HostClaim || !cfg.MovesEnabled {
+ if !cfg.HostClaim {
return false
}
alerting.Arm(alerting.FromEnv(ctx, getenv, nil))
diff --git a/cmd/console/movesswitch_test.go b/cmd/console/movesswitch_test.go
index 6173a520a..48730c28d 100644
--- a/cmd/console/movesswitch_test.go
+++ b/cmd/console/movesswitch_test.go
@@ -103,17 +103,17 @@ func isIdent(e ast.Expr, name string) bool {
// The console pages from its move pass (moves Plan 5 Task 7, delta M14):
// armMovesPaging arms alerting's sinks from the agent's variables on a
-// host-claim console with the moves switch on, and nowhere else; serve()
+// host-claim console, with or without moves; serve()
// calls it with os.Getenv before ConfigureClaimRuntime, so the recovery
// pass never runs unarmed.
-func TestArmMovesPagingOnlyWithBothSwitches(t *testing.T) {
+func TestArmMovesPagingWithClaims(t *testing.T) {
env := func(k string) string {
if k == "CARLOS_ALERT_WEBHOOK" {
return "http://example.test/hook"
}
return ""
}
- for _, tc := range []struct{ claim, moves, want bool }{{false, false, false}, {true, false, false}, {false, true, false}, {true, true, true}} {
+ for _, tc := range []struct{ claim, moves, want bool }{{false, false, false}, {true, false, true}, {false, true, false}, {true, true, true}} {
cfg := console.Config{HostClaim: tc.claim, MovesEnabled: tc.moves}
if got := armMovesPaging(context.Background(), cfg, env); got != tc.want {
t.Errorf("claim %v moves %v: armed %v, want %v", tc.claim, tc.moves, got, tc.want)
diff --git a/internal/console/claim_authority_census_test.go b/internal/console/claim_authority_census_test.go
index ee731d684..a0e03fae4 100644
--- a/internal/console/claim_authority_census_test.go
+++ b/internal/console/claim_authority_census_test.go
@@ -73,7 +73,11 @@ func claimAuthoritySites(t *testing.T) []string {
case *ast.Ident:
method = c.Name
}
- if !selected[method] {
+ // Loss paging reads signed control evidence and writes only its
+ // NonApp delivery journal. Inventory those sites explicitly.
+ lossPageSite := e.Name() == "claim_loss_pages.go" && (method == "List" || method == "ReadLossRecord" || method == "Exec" || method == "ExecContext" || method == "QueryRowContext" || method == "ClaimLossPageAttempt")
+ lossPageSchema := e.Name() == "store_gate.go" && fn.Name.Name == "OpenGated" && method == "Exec"
+ if !selected[method] && !lossPageSite && !lossPageSchema {
return true
}
var b bytes.Buffer
diff --git a/internal/console/claim_loss_pages.go b/internal/console/claim_loss_pages.go
new file mode 100644
index 000000000..75ef9b2be
--- /dev/null
+++ b/internal/console/claim_loss_pages.go
@@ -0,0 +1,166 @@
+package console
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "log"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/alerting"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostops"
+)
+
+const (
+ claimLossPageInterval = time.Hour
+ claimLossPageDailyInterval = 24 * time.Hour
+ claimLossPageBudget = 30 * time.Second
+)
+
+// claimLossReadBudget bounds one record's read inside a scan. A record whose
+// read keeps stalling then fails on its own and the scan moves past it,
+// retrying it next lap, instead of spending the whole scan budget on it every
+// time and starving the records behind it.
+var claimLossReadBudget = 5 * time.Second
+
+// Recovery schedules one bounded scan. A slow sink holds only this worker;
+// later recovery passes skip scheduling until it exits.
+func (s *Server) scheduleClaimLossPages(ctx context.Context) {
+ if !s.cfg.HostClaim || ctx.Err() != nil {
+ return
+ }
+ s.claimLossPageMu.Lock()
+ defer s.claimLossPageMu.Unlock()
+ if s.claimLossPageDone != nil {
+ select {
+ case <-s.claimLossPageDone:
+ default:
+ return
+ }
+ }
+ done := make(chan struct{})
+ s.claimLossPageDone = done
+ go func() {
+ defer close(done)
+ pass, cancel := context.WithTimeout(ctx, claimLossPageBudget)
+ defer cancel()
+ if err := s.pageClaimLosses(pass); err != nil {
+ log.Printf("console: claim recovery: %v", err)
+ }
+ }()
+}
+
+func (s *Server) pageClaimLosses(ctx context.Context) error {
+ return s.pageClaimLossesAt(ctx, time.Now)
+}
+
+// Scan records independently of claims: loss evidence outlives a holder or
+// operation, and may still need attention after its claim has gone.
+func (s *Server) pageClaimLossesAt(ctx context.Context, now func() time.Time) error {
+ if !s.cfg.HostClaim {
+ return nil
+ }
+ if s.cfg.Releases == nil || s.store == nil || len(s.cfg.ReleasePub) == 0 {
+ return hostops.ErrContract
+ }
+ var after string
+ if err := s.store.db.QueryRowContext(ctx, `SELECT last_key FROM claim_loss_pages WHERE rule = 'scan'`).Scan(&after); err != nil && !errors.Is(err, sql.ErrNoRows) {
+ return err
+ }
+ keys, err := s.cfg.Releases.List(ctx, "control/host-loss/")
+ if err != nil {
+ return err
+ }
+ // Store listings need not be ordered. Visit each key once, wrapping even
+ // when the saved key has since been removed or earlier keys were added.
+ sort.Strings(keys)
+ start := sort.Search(len(keys), func(i int) bool { return keys[i] > after })
+ var errs []error
+ for i := range keys {
+ if err := ctx.Err(); err != nil {
+ return errors.Join(append(errs, err)...)
+ }
+ key := keys[(start+i)%len(keys)]
+ errs = append(errs, s.pageClaimLossRecord(ctx, key, now))
+ if err := ctx.Err(); err != nil {
+ return errors.Join(append(errs, err)...)
+ }
+ // Never advance past canceled work. Other record errors are retried on
+ // the next lap so one bad record cannot block everything after it.
+ if _, err := s.store.db.ExecContext(ctx, `INSERT INTO claim_loss_pages (rule, first_attempted_at, attempted_at, last_key) VALUES ('scan', 0, 0, ?)
+ON CONFLICT(rule) DO UPDATE SET last_key = excluded.last_key`, key); err != nil {
+ return errors.Join(append(errs, err)...)
+ }
+ }
+ return errors.Join(errs...)
+}
+
+func (s *Server) pageClaimLossRecord(ctx context.Context, key string, now func() time.Time) error {
+ id := strings.TrimSuffix(strings.TrimPrefix(key, "control/host-loss/"), ".json")
+ exact, err := hostclaim.LossRecordKey(id)
+ if err != nil || exact != key {
+ return hostclaim.ErrLossRecord
+ }
+ read, cancel := context.WithTimeout(ctx, claimLossReadBudget)
+ observed, err := hostclaim.ReadLossRecord(read, s.cfg.Releases, s.cfg.ReleasePub, id)
+ cancel()
+ if errors.Is(err, hostclaim.ErrLossRecordAbsent) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ r := observed.Record
+ var errs []error
+ if r.AlertsPending {
+ subject, body := claimLossPendingPage(r.Host, r.ID)
+ errs = append(errs, s.pageClaimLoss(ctx, "pending", r, now, subject, body))
+ }
+ if r.Timers.Knowledge == hostclaim.TimersPartial || r.Timers.Knowledge == hostclaim.TimersUnknown {
+ subject, body := claimTimerLossPage(r.Host, r.ID)
+ errs = append(errs, s.pageClaimLoss(ctx, "timers", r, now, subject, body))
+ }
+ return errors.Join(errs...)
+}
+
+func (s *Server) pageClaimLoss(ctx context.Context, condition string, r hostclaim.LossRecord, now func() time.Time, subject, body string) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ attempted, err := s.store.ClaimLossPageAttempt(ctx, condition+" "+r.ID, now())
+ if err != nil || !attempted {
+ return err
+ }
+ return alerting.Page(ctx, alerting.LevelWarning, subject, body)
+}
+
+// ClaimLossPageAttempt reserves the attempt before delivery, including failed
+// sends. The conditional write also serializes overlapping recovery passes.
+func (s *Store) ClaimLossPageAttempt(ctx context.Context, rule string, now time.Time) (bool, error) {
+ result, err := s.db.ExecContext(ctx, `INSERT INTO claim_loss_pages (rule, first_attempted_at, attempted_at) VALUES (?, ?, ?)
+ON CONFLICT(rule) DO UPDATE SET attempted_at = excluded.attempted_at
+WHERE claim_loss_pages.attempted_at <= CASE
+WHEN claim_loss_pages.first_attempted_at > ? THEN ? ELSE ? END`,
+ rule, now.UnixNano(), now.UnixNano(), now.Add(-claimLossPageDailyInterval).UnixNano(),
+ now.Add(-claimLossPageInterval).UnixNano(), now.Add(-claimLossPageDailyInterval).UnixNano())
+ if err != nil {
+ return false, err
+ }
+ n, err := result.RowsAffected()
+ return n == 1, err
+}
+
+// Draft operator copy. The controller collects these strings for approval.
+func claimLossPendingPage(host, id string) (subject, body string) {
+ return fmt.Sprintf("Loss recorded for %s", host),
+ fmt.Sprintf("A loss was recorded for host %s. Review loss record %s and check what needs recovery.", host, id)
+}
+
+func claimTimerLossPage(host, id string) (subject, body string) {
+ return fmt.Sprintf("Timers may be lost for %s", host),
+ fmt.Sprintf("Timers may have been lost with the box serving host %s. Check loss record %s. Check which timers are missing and re-arm them.", host, id)
+}
diff --git a/internal/console/claim_loss_pages_test.go b/internal/console/claim_loss_pages_test.go
new file mode 100644
index 000000000..27771f3f7
--- /dev/null
+++ b/internal/console/claim_loss_pages_test.go
@@ -0,0 +1,730 @@
+package console
+
+import (
+ "context"
+ "crypto/ed25519"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "path/filepath"
+ "slices"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/service/sns"
+ "github.com/carlosframework/platform/internal/alerting"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func lossPageRecord(id string, pending bool, knowledge hostclaim.TimerKnowledge) hostclaim.LossRecord {
+ at := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
+ r := hostclaim.LossRecord{V: 1, ID: id, State: "prepared", Actor: "operator", AcceptedAt: at,
+ Host: "host.example", HolderIncarnation: strings.Repeat("a", 64), HolderStamp: 1,
+ ExpectedClaimGen: 1, ExpectedClaimETag: "claim", OpID: "operation", OpKind: hostclaim.KindMoving,
+ OpPhase: "precommit", Box: "box-a", Enrollment: "enrollment", QuarantineRevision: 1, FencedAt: at,
+ CloudObservation: hostclaim.LossCloudObservation{ObservedAt: at, Account: "123", Region: "eu-west-1", InstanceID: "i-1", State: "stopped", RequestID: "request"},
+ OperatorAttestation: hostclaim.LossOperatorAttestation{MaintainComputeFence: true, ColdQuarantineStartOnly: true},
+ ConfirmedConsequences: hostclaim.LossConsequences{RecoveryIntent: json.RawMessage(`{"kind":"lost-target"}`)},
+ AlertsPending: pending, Timers: hostclaim.TimerLoss{Knowledge: knowledge}}
+ if knowledge != hostclaim.TimersUnknown {
+ r.Timers.Source = &hostclaim.TimerSource{Bucket: "bucket", Key: "timers", VersionID: "v1", SHA256: strings.Repeat("b", 64), CapturedAt: at}
+ }
+ return r
+}
+
+func lossPageWorld(t *testing.T) (*Server, *release.DirStore, release.Signer, *[]map[string]string) {
+ t.Helper()
+ raw := release.NewDirStore(t.TempDir())
+ _, key, err := ed25519.GenerateKey(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ publishRestoreRoster(t, raw, key)
+ s := newEnabledRestoreConsole(t, raw, key, crConsole)
+ var pages []map[string]string
+ alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(_ string, body []byte) error {
+ var page map[string]string
+ if err := json.Unmarshal(body, &page); err != nil {
+ return err
+ }
+ pages = append(pages, page)
+ return nil
+ }})
+ t.Cleanup(func() { alerting.Arm(nil) })
+ return s, raw, release.RootSigner(key), &pages
+}
+
+func putLossPageRecord(t *testing.T, raw release.Store, signer release.Signer, r hostclaim.LossRecord) {
+ t.Helper()
+ if _, err := hostclaim.CreateLossRecord(t.Context(), raw, signer, r); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestClaimLossPagesIntervalAndRestart(t *testing.T) {
+ s, raw, signer, pages := lossPageWorld(t)
+ for _, id := range []string{"loss-one", "loss-two"} {
+ putLossPageRecord(t, raw, signer, lossPageRecord(id, true, hostclaim.TimersComplete))
+ }
+ now := time.Date(2026, 10, 12, 13, 0, 0, 123, time.UTC)
+ pass := func() {
+ t.Helper()
+ if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now }); err != nil {
+ t.Fatal(err)
+ }
+ }
+ pass()
+ pass()
+ if len(*pages) != 2 {
+ t.Fatalf("first passes: %v", *pages)
+ }
+ // Reopen the actual SQLite file; no in-memory limiter survives this.
+ path := s.store.gate.path
+ if err := s.store.Close(); err != nil {
+ t.Fatal(err)
+ }
+ st, err := OpenGated(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { st.Close() })
+ s = &Server{cfg: s.cfg, store: st}
+ now = now.Add(time.Hour - time.Nanosecond)
+ pass()
+ if len(*pages) != 2 {
+ t.Fatalf("restart lost rate limit: %v", *pages)
+ }
+ now = now.Add(time.Nanosecond)
+ pass()
+ if len(*pages) != 4 {
+ t.Fatalf("interval did not repeat each record: %v", *pages)
+ }
+ key, _ := hostclaim.LossRecordKey("loss-one")
+ if err := raw.Delete(t.Context(), key); err != nil {
+ t.Fatal(err)
+ }
+ now = now.Add(time.Hour)
+ pass()
+ if len(*pages) != 5 || !strings.Contains((*pages)[4]["body"], "loss-two") {
+ t.Fatalf("absent pending record paged: %v", *pages)
+ }
+}
+
+func TestClaimLossPagesClearedAndTimers(t *testing.T) {
+ for _, knowledge := range []hostclaim.TimerKnowledge{hostclaim.TimersComplete, hostclaim.TimersPartial, hostclaim.TimersUnknown} {
+ t.Run(string(knowledge), func(t *testing.T) {
+ s, raw, signer, pages := lossPageWorld(t)
+ r := lossPageRecord("loss-timer", false, knowledge)
+ putLossPageRecord(t, raw, signer, r)
+ now := r.AcceptedAt
+ if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now }); err != nil {
+ t.Fatal(err)
+ }
+ want := 1
+ if knowledge == hostclaim.TimersComplete {
+ want = 0
+ }
+ if len(*pages) != want {
+ t.Fatalf("pages: %v", *pages)
+ }
+ if want != 0 && (!strings.Contains((*pages)[0]["body"], r.Host) || !strings.Contains((*pages)[0]["body"], "re-arm")) {
+ t.Fatalf("timer instructions: %v", *pages)
+ }
+ key, _ := hostclaim.LossRecordKey(r.ID)
+ if err := raw.Delete(t.Context(), key); err != nil {
+ t.Fatal(err)
+ }
+ if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now.Add(2 * time.Hour) }); err != nil {
+ t.Fatal(err)
+ }
+ if len(*pages) != want {
+ t.Fatal("cleared record paged")
+ }
+ })
+ }
+}
+
+func TestClaimLossPagesRecoveryWiringWithoutMovesOrClaims(t *testing.T) {
+ s, raw, signer, pages := lossPageWorld(t)
+ putLossPageRecord(t, raw, signer, lossPageRecord("orphan-loss", true, hostclaim.TimersComplete))
+ if s.cfg.MovesEnabled {
+ t.Fatal("fixture enabled moves")
+ }
+ if err := s.RunClaimRecoveryPass(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ waitClaimLossPages(t, s)
+ if len(*pages) != 1 {
+ t.Fatalf("record without a claim was not paged: %v", *pages)
+ }
+}
+
+func TestClaimLossPagesFlagOff(t *testing.T) {
+ calls := 0
+ alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(string, []byte) error { calls++; return nil }})
+ t.Cleanup(func() { alerting.Arm(nil) })
+ s := &Server{}
+ if err := s.pageClaimLossesAt(t.Context(), time.Now); err != nil {
+ t.Fatal(err)
+ }
+ if err := s.RunClaimRecoveryPass(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ s.scheduleClaimLossPages(t.Context())
+ if s.claimLossPageDone != nil {
+ t.Fatal("disabled runtime started a worker")
+ }
+ if calls != 0 {
+ t.Fatal("disabled runtime paged")
+ }
+ st, err := Open(filepath.Join(t.TempDir(), "off.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer st.Close()
+ var count int
+ if err := st.db.QueryRow(`SELECT count(*) FROM sqlite_master WHERE name = 'claim_loss_pages'`).Scan(&count); err != nil || count != 0 {
+ t.Fatalf("flag-off schema changed: %d %v", count, err)
+ }
+}
+
+func TestClaimLossPageFailedDeliveryIsRateLimited(t *testing.T) {
+ s, raw, signer, _ := lossPageWorld(t)
+ putLossPageRecord(t, raw, signer, lossPageRecord("failed-page", true, hostclaim.TimersComplete))
+ calls := 0
+ failure := errors.New("delivery failed")
+ alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(string, []byte) error { calls++; return failure }})
+ now := time.Now()
+ if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now }); !errors.Is(err, failure) {
+ t.Fatalf("delivery: %v", err)
+ }
+ if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now.Add(time.Minute) }); err != nil {
+ t.Fatal(err)
+ }
+ if calls != 1 {
+ t.Fatalf("attempts: %d", calls)
+ }
+}
+
+func TestClaimLossPagesAutomaticReportingDoesNotAcknowledge(t *testing.T) {
+ s, raw, _, pages := lossPageWorld(t)
+ ref := hostclaim.FenceRef{Enrollment: strings.Repeat("a", 32), SHA256: strings.Repeat("b", 64), Key: "control/box-fences/" + strings.Repeat("a", 32) + ".json"}
+ sum := sha256.Sum256([]byte("operation\x00box-a\x00" + ref.Enrollment + "\x00" + ref.SHA256))
+ id := "auto-" + hex.EncodeToString(sum[:16])
+ loss := hostclaim.AutomaticLoss{ID: id, Box: "box-a", Enrollment: ref.Enrollment, Fence: ref,
+ ObservedAt: time.Now().UTC().Format(time.RFC3339Nano), Op: "operation", Kind: hostclaim.KindLanding,
+ Phase: hostclaim.PhaseProbing, Role: "target", Incarnation: strings.Repeat("c", 64), Stamp: 1, Timers: hostclaim.TimersUnknown}
+ _, err := hostclaim.Update(t.Context(), raw, hostclaim.Observed{Host: "host.example", Absent: true}, func(c *hostclaim.Claim) error {
+ c.LossRefs = []string{id}
+ c.EvidenceRefs = []hostclaim.AutomaticLoss{loss}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := s.RunClaimRecoveryPass(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ waitClaimLossPages(t, s)
+ if len(*pages) != 2 {
+ t.Fatalf("materialized record must page both conditions in the same pass: %v", *pages)
+ }
+ c, err := hostclaim.Read(t.Context(), raw, "host.example")
+ if err != nil || !c.Claim.EvidenceRefs[0].Reported {
+ t.Fatalf("reporting did not finish: %+v %v", c, err)
+ }
+ if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return time.Now().Add(2 * time.Hour) }); err != nil {
+ t.Fatal(err)
+ }
+ if len(*pages) != 4 {
+ t.Fatal("claim reporting silenced pending operator alerts")
+ }
+}
+
+func TestClaimLossPageReservationSerializesPasses(t *testing.T) {
+ st := openGatedStore(t)
+ now := time.Now()
+ var wg sync.WaitGroup
+ var attempts atomic.Int64
+ for i := 0; i < 12; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ ok, err := st.ClaimLossPageAttempt(t.Context(), "pending loss", now)
+ if err != nil {
+ t.Error(err)
+ }
+ if ok {
+ attempts.Add(1)
+ }
+ }()
+ }
+ wg.Wait()
+ if attempts.Load() != 1 {
+ t.Fatalf("reserved %d attempts", attempts.Load())
+ }
+}
+
+func TestClaimLossPagesRejectBadEvidenceAndContinue(t *testing.T) {
+ s, raw, signer, pages := lossPageWorld(t)
+ if _, err := raw.Put(t.Context(), "control/host-loss/bad.json", []byte(`{}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ putLossPageRecord(t, raw, signer, lossPageRecord("valid", true, hostclaim.TimersComplete))
+ if err := s.pageClaimLossesAt(t.Context(), time.Now); !errors.Is(err, hostclaim.ErrLossRecord) {
+ t.Fatalf("invalid signature accepted: %v", err)
+ }
+ if len(*pages) != 1 {
+ t.Fatalf("bad record blocked good record: %v", *pages)
+ }
+}
+
+func TestClaimLossPagesJournalFailureDoesNotSend(t *testing.T) {
+ s, raw, signer, pages := lossPageWorld(t)
+ putLossPageRecord(t, raw, signer, lossPageRecord("loss", true, hostclaim.TimersComplete))
+ if err := s.store.Close(); err != nil {
+ t.Fatal(err)
+ }
+ if err := s.pageClaimLossesAt(t.Context(), time.Now); err == nil {
+ t.Fatal("journal error hidden")
+ }
+ if len(*pages) != 0 {
+ t.Fatal("sent without durable reservation")
+ }
+}
+
+func waitClaimLossPages(t *testing.T, s *Server) {
+ t.Helper()
+ s.claimLossPageMu.Lock()
+ done := s.claimLossPageDone
+ s.claimLossPageMu.Unlock()
+ if done == nil {
+ t.Fatal("no loss paging worker scheduled")
+ }
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("loss paging worker did not finish")
+ }
+}
+
+func TestClaimLossPagesHourlyToDailyAcrossReopen(t *testing.T) {
+ s, raw, signer, pages := lossPageWorld(t)
+ r := lossPageRecord("old-loss", true, hostclaim.TimersUnknown)
+ putLossPageRecord(t, raw, signer, r)
+ first := r.AcceptedAt.Add(30 * 24 * time.Hour)
+ pass := func(at time.Time, want int) {
+ t.Helper()
+ if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return at }); err != nil {
+ t.Fatal(err)
+ }
+ if len(*pages) != want {
+ t.Fatalf("at %s: got %d pages, want %d", at.Sub(first), len(*pages), want)
+ }
+ }
+ for hour := 0; hour < 24; hour++ {
+ pass(first.Add(time.Duration(hour)*time.Hour), 2*(hour+1))
+ }
+ path := s.store.gate.path
+ if err := s.store.Close(); err != nil {
+ t.Fatal(err)
+ }
+ st, err := OpenGated(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { st.Close() })
+ s = &Server{cfg: s.cfg, store: st}
+ pass(first.Add(24*time.Hour-time.Nanosecond), 48)
+ // Once the first 24 hours end, the last attempt must be a day old.
+ pass(first.Add(24*time.Hour), 48)
+ pass(first.Add(47*time.Hour-time.Nanosecond), 48)
+ pass(first.Add(47*time.Hour), 50)
+ pass(first.Add(71*time.Hour-time.Nanosecond), 50)
+ pass(first.Add(71*time.Hour), 52)
+ for _, condition := range []string{"pending", "timers"} {
+ var saved int64
+ if err := st.db.QueryRow(`SELECT first_attempted_at FROM claim_loss_pages WHERE rule = ?`, condition+" "+r.ID).Scan(&saved); err != nil || saved != first.UnixNano() {
+ t.Fatalf("first %s attempt changed: %d %v", condition, saved, err)
+ }
+ }
+}
+
+func TestClaimLossPagesClockAtEachReservation(t *testing.T) {
+ s, raw, signer, _ := lossPageWorld(t)
+ for _, id := range []string{"a-early", "b-late"} {
+ putLossPageRecord(t, raw, signer, lossPageRecord(id, true, hostclaim.TimersComplete))
+ }
+ first := time.Now()
+ now := first
+ late := 0
+ calls := 0
+ alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(_ string, b []byte) error {
+ calls++
+ if strings.Contains(string(b), "b-late") {
+ late++
+ }
+ if calls == 1 {
+ now = now.Add(time.Hour)
+ }
+ return nil
+ }})
+ clock := func() time.Time { return now }
+ if err := s.pageClaimLossesAt(t.Context(), clock); err != nil {
+ t.Fatal(err)
+ }
+ now = now.Add(10 * time.Second)
+ if err := s.pageClaimLossesAt(t.Context(), clock); err != nil {
+ t.Fatal(err)
+ }
+ if late != 1 || calls != 3 {
+ t.Fatalf("late record retried ten seconds later: late=%d, total=%d", late, calls)
+ }
+ var reserved int64
+ if err := s.store.db.QueryRow(`SELECT attempted_at FROM claim_loss_pages WHERE rule = 'pending b-late'`).Scan(&reserved); err != nil || reserved != first.Add(time.Hour).UnixNano() {
+ t.Fatalf("late reservation: %d %v", reserved, err)
+ }
+}
+
+func TestClaimLossPagesBlockedSinkDoesNotBlockRecovery(t *testing.T) {
+ s, raw, signer, _ := lossPageWorld(t)
+ putLossPageRecord(t, raw, signer, lossPageRecord("blocking", true, hostclaim.TimersComplete))
+ entered, releaseSink := make(chan struct{}), make(chan struct{})
+ var unblock sync.Once
+ var calls atomic.Int64
+ alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(string, []byte) error {
+ if calls.Add(1) == 1 {
+ close(entered)
+ <-releaseSink
+ }
+ return nil
+ }})
+ t.Cleanup(func() {
+ unblock.Do(func() { close(releaseSink) })
+ waitClaimLossPages(t, s)
+ })
+ var records, purges, admissions atomic.Int64
+ wrap := func(fn func(context.Context) error, count *atomic.Int64) func(context.Context) error {
+ return func(ctx context.Context) error { count.Add(1); return fn(ctx) }
+ }
+ r := s.cfg.HostClaimRecovery
+ r.SweepRecords = wrap(r.SweepRecords, &records)
+ r.SweepPurges = wrap(r.SweepPurges, &purges)
+ r.SweepAdmissions = wrap(r.SweepAdmissions, &admissions)
+ pass := func() {
+ t.Helper()
+ done := make(chan error, 1)
+ go func() { done <- s.RunClaimRecoveryPass(t.Context()) }()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(2 * time.Second):
+ unblock.Do(func() { close(releaseSink) })
+ <-done
+ t.Fatal("paging blocked recovery")
+ }
+ }
+ pass()
+ select {
+ case <-entered:
+ case <-time.After(2 * time.Second):
+ t.Fatal("sink was not called")
+ }
+ s.claimLossPageMu.Lock()
+ worker := s.claimLossPageDone
+ s.claimLossPageMu.Unlock()
+ for range 3 {
+ pass()
+ }
+ if records.Load() != 4 || purges.Load() != 4 || admissions.Load() != 4 {
+ t.Fatalf("sweeps delayed: records=%d purges=%d admissions=%d", records.Load(), purges.Load(), admissions.Load())
+ }
+ s.claimLossPageMu.Lock()
+ overlapped := s.claimLossPageDone != worker
+ s.claimLossPageMu.Unlock()
+ if overlapped || calls.Load() != 1 {
+ t.Fatal("paging workers overlapped")
+ }
+ unblock.Do(func() { close(releaseSink) })
+ waitClaimLossPages(t, s)
+ putLossPageRecord(t, raw, signer, lossPageRecord("next-worker", true, hostclaim.TimersComplete))
+ pass()
+ waitClaimLossPages(t, s)
+ if calls.Load() != 2 {
+ t.Fatal("next recovery did not schedule a new paging worker")
+ }
+}
+
+func TestClaimLossPageJournalUpgrade(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "old.db")
+ st, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ at := time.Now().Add(-23 * time.Hour)
+ _, err = st.db.Exec(`CREATE TABLE claim_loss_pages (rule TEXT PRIMARY KEY, attempted_at INTEGER NOT NULL);
+INSERT INTO claim_loss_pages VALUES ('pending old', ?)`, at.UnixNano())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := st.Close(); err != nil {
+ t.Fatal(err)
+ }
+ for range 2 {
+ st, err := OpenGated(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var first, last int64
+ err = st.db.QueryRow(`SELECT first_attempted_at, attempted_at FROM claim_loss_pages WHERE rule = 'pending old'`).Scan(&first, &last)
+ var cursor string
+ cursorErr := st.db.QueryRow(`SELECT last_key FROM claim_loss_pages WHERE rule = 'pending old'`).Scan(&cursor)
+ st.Close()
+ if err != nil || first != at.UnixNano() || last != at.UnixNano() {
+ t.Fatalf("upgrade lost durable attempt: %d %d %v", first, last, err)
+ }
+ if cursorErr != nil || cursor != "" {
+ t.Fatalf("upgrade scan progress: %q %v", cursor, cursorErr)
+ }
+ }
+}
+
+type lossPageDeadlineStore struct {
+ release.Store
+ deadline chan time.Time
+}
+
+func (s lossPageDeadlineStore) List(ctx context.Context, prefix string) ([]string, error) {
+ deadline, _ := ctx.Deadline()
+ s.deadline <- deadline
+ <-ctx.Done()
+ return nil, ctx.Err()
+}
+
+func TestClaimLossPagesWorkerBudget(t *testing.T) {
+ s, _, _, pages := lossPageWorld(t)
+ deadline := make(chan time.Time, 1)
+ s.cfg.Releases = lossPageDeadlineStore{Store: s.cfg.Releases, deadline: deadline}
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ before := time.Now()
+ s.scheduleClaimLossPages(ctx)
+ select {
+ case at := <-deadline:
+ if at.Before(before.Add(claimLossPageBudget)) || at.After(time.Now().Add(claimLossPageBudget)) {
+ t.Errorf("scan has no per-pass budget: deadline=%v", at)
+ }
+ case <-time.After(time.Second):
+ t.Error("worker never started discovery")
+ }
+ cancel()
+ waitClaimLossPages(t, s)
+ if len(*pages) != 0 {
+ t.Fatal("canceled scan delivered a page")
+ }
+ s.claimLossPageMu.Lock()
+ done := s.claimLossPageDone
+ s.claimLossPageMu.Unlock()
+ s.scheduleClaimLossPages(ctx)
+ if s.claimLossPageDone != done {
+ t.Fatal("canceled recovery scheduled another worker")
+ }
+}
+
+type lossPageBlockingSNS struct {
+ entered chan time.Time
+}
+
+func (p lossPageBlockingSNS) Publish(ctx context.Context, _ *sns.PublishInput, _ ...func(*sns.Options)) (*sns.PublishOutput, error) {
+ deadline, _ := ctx.Deadline()
+ p.entered <- deadline
+ <-ctx.Done()
+ return nil, ctx.Err()
+}
+
+func TestClaimLossPagesSinkTimeoutAndCancellation(t *testing.T) {
+ s, raw, signer, _ := lossPageWorld(t)
+ putLossPageRecord(t, raw, signer, lossPageRecord("blocked-sns", true, hostclaim.TimersUnknown))
+ entered := make(chan time.Time, 1)
+ alerting.Arm(&alerting.Sinks{TopicARN: "test", Pub: lossPageBlockingSNS{entered: entered}})
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ s.scheduleClaimLossPages(ctx)
+ select {
+ case deadline := <-entered:
+ if deadline.IsZero() || time.Until(deadline) > alerting.PublishTimeout {
+ t.Errorf("SNS sink has no timeout: %v", deadline)
+ }
+ case <-time.After(time.Second):
+ t.Error("worker never called SNS")
+ }
+ cancel()
+ waitClaimLossPages(t, s)
+ var attempts int
+ if err := s.store.db.QueryRow(`SELECT count(*) FROM claim_loss_pages`).Scan(&attempts); err != nil || attempts != 1 {
+ t.Fatalf("canceled pass reserved another condition: %d %v", attempts, err)
+ }
+}
+
+type lossPageSlowStore struct {
+ release.Store
+ reads []string
+ reverse bool
+}
+
+func (s *lossPageSlowStore) List(ctx context.Context, prefix string) ([]string, error) {
+ keys, err := s.Store.List(ctx, prefix)
+ if s.reverse {
+ slices.Reverse(keys)
+ }
+ return keys, err
+}
+
+func (s *lossPageSlowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.reads = append(s.reads, key)
+ if len(s.reads) > 2 {
+ <-ctx.Done()
+ return nil, "", ctx.Err()
+ }
+ timer := time.NewTimer(10 * time.Millisecond)
+ defer timer.Stop()
+ select {
+ case <-timer.C:
+ return s.Store.Get(ctx, key)
+ case <-ctx.Done():
+ return nil, "", ctx.Err()
+ }
+}
+
+func TestClaimLossPagesScanProgressSurvivesRestartAndWraps(t *testing.T) {
+ s, raw, signer, pages := lossPageWorld(t)
+ for _, id := range []string{"b-retained", "c-retained", "d-retained", "e-retained", "f-retained", "z-pending"} {
+ putLossPageRecord(t, raw, signer, lossPageRecord(id, id == "z-pending", hostclaim.TimersComplete))
+ }
+ slow := &lossPageSlowStore{Store: raw}
+ s.cfg.Releases = slow
+ pass := func(first string) {
+ t.Helper()
+ slow.reads = nil
+ ctx, cancel := context.WithTimeout(t.Context(), 250*time.Millisecond)
+ defer cancel()
+ if err := s.pageClaimLossesAt(ctx, time.Now); !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("slow scan: %v", err)
+ }
+ key, _ := hostclaim.LossRecordKey(first)
+ if len(slow.reads) != 3 || slow.reads[0] != key {
+ t.Fatalf("scan reads: %v; want first %s and one unfinished read", slow.reads, key)
+ }
+ }
+ pass("b-retained")
+ if len(*pages) != 0 {
+ t.Fatalf("first scan reached beyond its budget: %v", *pages)
+ }
+ path := s.store.gate.path
+ if err := s.store.Close(); err != nil {
+ t.Fatal(err)
+ }
+ st, err := OpenGated(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { st.Close() })
+ s = &Server{cfg: s.cfg, store: st}
+ pass("d-retained")
+ pass("f-retained")
+ if len(*pages) != 1 || !strings.Contains((*pages)[0]["body"], "z-pending") {
+ t.Fatalf("pending record not delivered within three scans: %v", *pages)
+ }
+ putLossPageRecord(t, raw, signer, lossPageRecord("a-new", true, hostclaim.TimersComplete))
+ key, _ := hostclaim.LossRecordKey("z-pending")
+ if err := raw.Delete(t.Context(), key); err != nil {
+ t.Fatal(err)
+ }
+ slow.reverse = true
+ pass("a-new")
+ if len(*pages) != 2 || !strings.Contains((*pages)[1]["body"], "a-new") {
+ t.Fatalf("scan did not wrap past the removed cursor to the new early record: %v", *pages)
+ }
+}
+
+func TestClaimLossPagesScanRetriesUnfinishedPendingRead(t *testing.T) {
+ s, raw, signer, pages := lossPageWorld(t)
+ for _, id := range []string{"a-retained", "b-retained", "c-pending"} {
+ putLossPageRecord(t, raw, signer, lossPageRecord(id, id == "c-pending", hostclaim.TimersComplete))
+ }
+ s.cfg.Releases = &lossPageSlowStore{Store: raw}
+ ctx, cancel := context.WithTimeout(t.Context(), 250*time.Millisecond)
+ defer cancel()
+ if err := s.pageClaimLossesAt(ctx, time.Now); !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("slow scan: %v", err)
+ }
+ if len(*pages) != 0 {
+ t.Fatalf("unfinished read paged: %v", *pages)
+ }
+ slow := &lossPageSlowStore{Store: raw}
+ s.cfg.Releases = slow
+ ctx, cancel = context.WithTimeout(t.Context(), 250*time.Millisecond)
+ defer cancel()
+ if err := s.pageClaimLossesAt(ctx, time.Now); !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("resumed scan: %v", err)
+ }
+ key, _ := hostclaim.LossRecordKey("c-pending")
+ if !slices.Equal(slow.reads, []string{key, "control/host-loss/a-retained.json", "control/host-loss/b-retained.json"}) || len(*pages) != 1 {
+ t.Fatalf("unfinished pending read skipped: reads=%v pages=%v", slow.reads, *pages)
+ }
+}
+
+// lossPageStalledStore never answers a read of one key.
+type lossPageStalledStore struct {
+ release.Store
+ stalled string
+ reads []string
+}
+
+func (s *lossPageStalledStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.reads = append(s.reads, key)
+ if key == s.stalled {
+ <-ctx.Done()
+ return nil, "", ctx.Err()
+ }
+ return s.Store.Get(ctx, key)
+}
+
+func TestClaimLossPagesPersistentlyStalledReadDoesNotStarve(t *testing.T) {
+ prior := claimLossReadBudget
+ claimLossReadBudget = 50 * time.Millisecond
+ t.Cleanup(func() { claimLossReadBudget = prior })
+ s, raw, signer, pages := lossPageWorld(t)
+ for _, id := range []string{"a-stalled", "b-pending"} {
+ putLossPageRecord(t, raw, signer, lossPageRecord(id, true, hostclaim.TimersComplete))
+ }
+ stalledKey, _ := hostclaim.LossRecordKey("a-stalled")
+ pendingKey, _ := hostclaim.LossRecordKey("b-pending")
+ store := &lossPageStalledStore{Store: raw, stalled: stalledKey}
+ s.cfg.Releases = store
+ for lap := 0; lap < 2; lap++ {
+ store.reads = nil
+ ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
+ err := s.pageClaimLossesAt(ctx, time.Now)
+ cancel()
+ if err == nil || !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("lap %d: the stalled read should surface as a record error, got %v", lap, err)
+ }
+ if !slices.Contains(store.reads, pendingKey) {
+ t.Fatalf("lap %d: the record behind the stalled one was never read: %v", lap, store.reads)
+ }
+ if !slices.Contains(store.reads, stalledKey) {
+ t.Fatalf("lap %d: the stalled record was not retried: %v", lap, store.reads)
+ }
+ }
+ if len(*pages) != 1 || !strings.Contains((*pages)[0]["body"], "b-pending") {
+ t.Fatalf("pending record behind a stalled read was not paged exactly once: %v", *pages)
+ }
+}
diff --git a/internal/console/claim_runtime.go b/internal/console/claim_runtime.go
index 186476dfd..75f3f0bc0 100644
--- a/internal/console/claim_runtime.go
+++ b/internal/console/claim_runtime.go
@@ -127,6 +127,7 @@ func (s *Server) ConfigureClaimRuntime(ctx context.Context) error {
moveLogs = s.drainMoveLog
}
s.cfg.HostClaimRecovery = &hostops.Recovery{
+ ScheduleLossPages: s.scheduleClaimLossPages,
ReportAutomaticLosses: func(ctx context.Context, host string) error {
return a.Control.ReportAutomaticLosses(ctx, a.Signer, host)
},
diff --git a/internal/console/server.go b/internal/console/server.go
index 7a60ea992..da8ae45bd 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -821,6 +821,8 @@ type Config struct {
// Server is the console's http.Handler.
type Server struct {
+ claimLossPageMu sync.Mutex
+ claimLossPageDone chan struct{}
// Enabled claim runtime (ConfigureClaimRuntime). Unset flag-off.
claimRuntimeConfigured bool
claimOwner hostops.RegistrationOwner
diff --git a/internal/console/store_gate.go b/internal/console/store_gate.go
index 6a90ecb46..bda729a56 100644
--- a/internal/console/store_gate.go
+++ b/internal/console/store_gate.go
@@ -209,6 +209,7 @@ var claimSQLTables = map[string]claimSQLTable{
"memberships": {NonApp: "account membership, keyed by account and address; names no app"},
"invites": {NonApp: "account invites, keyed by account and address; names no app"},
"alert_seen": {NonApp: "operator alert memory, keyed by box, rule and address"},
+ "claim_loss_pages": {NonApp: "claim recovery page attempts per loss record and condition, and scan progress; no app state or lifecycle authority"},
"move_pages": {NonApp: "the move pass's last page attempt per finding (rule names the host and the move); no app column"},
"move_log_outbox": {NonApp: "a moves audit outbox that stores no app name: rows are keyed by host, move and entry and bound to an instance incarnation, verified against the host's claim and record at every write, written only by the moves recovery and log passes, and purged with the incarnation"},
"edge_proof_seen": {NonApp: "single-row edge proof marker"},
@@ -1789,6 +1790,24 @@ func OpenGated(path string) (*Store, error) {
if err := s.migrate(); err != nil {
return err
}
+ if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS claim_loss_pages (
+ rule TEXT PRIMARY KEY,
+ first_attempted_at INTEGER NOT NULL,
+ attempted_at INTEGER NOT NULL
+);`); err != nil {
+ return err
+ }
+ // The earlier journal retained only the last attempt. Preserve that
+ // durable time as the start for existing rows; record age is irrelevant.
+ if _, err := s.db.Exec(`ALTER TABLE claim_loss_pages ADD COLUMN first_attempted_at INTEGER`); err != nil && !strings.Contains(err.Error(), "duplicate column") {
+ return err
+ }
+ if _, err := s.db.Exec(`UPDATE claim_loss_pages SET first_attempted_at = attempted_at WHERE first_attempted_at IS NULL`); err != nil {
+ return err
+ }
+ if _, err := s.db.Exec(`ALTER TABLE claim_loss_pages ADD COLUMN last_key TEXT NOT NULL DEFAULT ''`); err != nil && !strings.Contains(err.Error(), "duplicate column") {
+ return err
+ }
return s.migrateClaimAppIdentity()
})
}
diff --git a/internal/console/testdata/claim-authority-sites.txt b/internal/console/testdata/claim-authority-sites.txt
index 9e1fbf855..8544c25fd 100644
--- a/internal/console/testdata/claim-authority-sites.txt
+++ b/internal/console/testdata/claim-authority-sites.txt
@@ -125,3 +125,13 @@ store_gate.go|openGated|sql.OpenDB(claimConnector{g: g, trusted: true})
store_gate.go|openGated|sql.OpenDB(claimConnector{g: g})
store_gate.go|withClaimSQL|s.withPreparedClaimSQL(ctx, scopes, check, func(ctx context.Context, _ claimSQLScope) error { return ctx.Err() }, nil, false, fn)
view.go|openRegistry|registry.Open(path)
+claim_loss_pages.go|ClaimLossPageAttempt|s.db.ExecContext(ctx, `INSERT INTO claim_loss_pages (rule, first_attempted_at, attempted_at) VALUES (?, ?, ?) ON CONFLICT(rule) DO UPDATE SET attempted_at = excluded.attempted_at WHERE claim_loss_pages.attempted_at <= CASE WHEN claim_loss_pages.first_attempted_at > ? THEN ? ELSE ? END`, rule, now.UnixNano(), now.UnixNano(), now.Add(-claimLossPageDailyInterval).UnixNano(), now.Add(-claimLossPageInterval).UnixNano(), now.Add(-claimLossPageDailyInterval).UnixNano())
+claim_loss_pages.go|pageClaimLossRecord|hostclaim.ReadLossRecord(read, s.cfg.Releases, s.cfg.ReleasePub, id)
+claim_loss_pages.go|pageClaimLossesAt|s.cfg.Releases.List(ctx, "control/host-loss/")
+claim_loss_pages.go|pageClaimLoss|s.store.ClaimLossPageAttempt(ctx, condition+" "+r.ID, now())
+store_gate.go|OpenGated|s.db.Exec(`CREATE TABLE IF NOT EXISTS claim_loss_pages ( rule TEXT PRIMARY KEY, first_attempted_at INTEGER NOT NULL, attempted_at INTEGER NOT NULL );`)
+store_gate.go|OpenGated|s.db.Exec(`ALTER TABLE claim_loss_pages ADD COLUMN first_attempted_at INTEGER`)
+store_gate.go|OpenGated|s.db.Exec(`UPDATE claim_loss_pages SET first_attempted_at = attempted_at WHERE first_attempted_at IS NULL`)
+claim_loss_pages.go|pageClaimLossesAt|s.store.db.ExecContext(ctx, `INSERT INTO claim_loss_pages (rule, first_attempted_at, attempted_at, last_key) VALUES ('scan', 0, 0, ?) ON CONFLICT(rule) DO UPDATE SET last_key = excluded.last_key`, key)
+claim_loss_pages.go|pageClaimLossesAt|s.store.db.QueryRowContext(ctx, `SELECT last_key FROM claim_loss_pages WHERE rule = 'scan'`)
+store_gate.go|OpenGated|s.db.Exec(`ALTER TABLE claim_loss_pages ADD COLUMN last_key TEXT NOT NULL DEFAULT ''`)
diff --git a/internal/hostops/recovery.go b/internal/hostops/recovery.go
index 32b6735ac..64d02ef1d 100644
--- a/internal/hostops/recovery.go
+++ b/internal/hostops/recovery.go
@@ -38,6 +38,8 @@ func ListClaims(ctx context.Context, s release.Store) ([]string, error) {
}
type Recovery struct {
+ // ScheduleLossPages triggers a separate worker after materialization.
+ ScheduleLossPages func(context.Context)
ReportAutomaticLosses func(context.Context, string) error
MoveFenceAuthority hostclaim.FenceAuthority
Moves func(context.Context, *MoveRecoveryGrant) error
@@ -121,6 +123,9 @@ func (r Recovery) Pass(ctx context.Context) error {
all = append(all, err)
}
}
+ if r.ScheduleLossPages != nil {
+ r.ScheduleLossPages(ctx)
+ }
if r.MoveLogs != nil {
if err := runMoveLogDrain(ctx, r.MoveLogs); err != nil && !errors.Is(err, ErrPending) {
all = append(all, err)
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index 519bb1e68..fed065e54 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -472,6 +472,8 @@ func Reviewed(ctx context.Context, s Store, k string) error {
// BEGIN REVIEWED INVENTORY
var plan9B8CensusInventory = []censusEntry{
+ // Read-only discovery of signed loss records; no record write or lifecycle authority.
+ {"internal/console/claim_loss_pages.go", "Server.pageClaimLossRecord", "key-site", "key:LossRecordKey", "d21e2450f540edf6645c457eb08c72f470969f5ea4fcbc597157bb7dbeb27287"},
{"internal/hostbox/parkleases.go", "ReplicaParkLeaseSet", "key-site", "lit:instances", "0ca8d9b743402fda9d15cc215464423b35e21c1c2b47f91294d76188b7d43fb6"},
{"internal/host/claimretirementcensus.go", "ClaimRetirementCensus", "key-site", "lit:instances", "abc0bd8c5f5d328abfac8821f42f8f23954723ed6506747f33b0197824e2611f"},
{"internal/host/claimproduction.go", "productionClaimRegional.View", "key-site", "lit:instances", "533203b2aa729801438da4f3a85799a58ea0e67d20eb5625f0de32ea623f1607"},
diff --git a/testdata/claim-moves-copy.md b/testdata/claim-moves-copy.md
index 6038aa4cb..179c4560b 100644
--- a/testdata/claim-moves-copy.md
+++ b/testdata/claim-moves-copy.md
@@ -1,7 +1,16 @@
-# Claim moves copy record (HC5-N1 approved by Paul 2026-10-08; HC5-N2 and HC5-N3, from claim Plan 10, approved by Paul 2026-10-10)
+# Claim moves copy record (HC5-N1 approved by Paul 2026-10-08; HC5-N2 and HC5-N3, from claim Plan 10, approved by Paul 2026-10-10; HC5-N4 to HC5-N7, claim loss pages, approved by Paul 2026-10-12)
| id | file | text | status |
|---|---|---|---|
| HC5-N1 | internal/console/claim_retire_helpers.go | instance moves are not available on this console yet: they need the host-claim Moves protocol | approved |
| HC5-N2 | internal/console/apiinstancemove.go | invalid resolution request | approved |
| HC5-N3 | internal/console/apiinstancemove.go | could not read the instance | approved |
+
+## Claim loss pages (pending Paul's review)
+
+| id | file | text | status |
+|---|---|---|---|
+| HC5-N4 | internal/console/claim_loss_pages.go | Loss recorded for <host> | approved |
+| HC5-N5 | internal/console/claim_loss_pages.go | A loss was recorded for host <host>. Review loss record <id> and check what needs recovery. | approved |
+| HC5-N6 | internal/console/claim_loss_pages.go | Timers may be lost for <host> | approved |
+| HC5-N7 | internal/console/claim_loss_pages.go | Timers may have been lost with the box serving host <host>. Check loss record <id>. Check which timers are missing and re-arm them. | approved |