amadan

carlos / platform Public

console: geoip doors (upload then job, reconciled verdict keyed by namespace, off), job results, staging dir, daily sweep/expiry; edge mirror re-reads the pointer before install. spec rev 9 (Astra round 8)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude ca3f35367bdb484fd116ec08e76ec60c03f5b8f4 parent 58beffa
12 files changed, +928 −14
  • cmd/console/main.go +6 −0
  • docs/superpowers/specs/2026-10-06-platform-geoip-design.md +45 −8
  • docs/superpowers/specs/CODEX-FINDINGS-GEOIP-R7.md +41 −0
  • docs/superpowers/specs/CODEX-FINDINGS-GEOIP-R8.md +37 −0
  • internal/console/apisystemgeoip.go +338 −0
  • internal/console/apisystemgeoip_test.go +300 −0
  • internal/console/apisystemupdate.go +43 −2
  • internal/console/authcli.go +5 −0
  • internal/console/server.go +38 −2
  • internal/console/svcroutes_test.go +5 −0
  • internal/edge/geoipmirror.go +33 −2
  • internal/edge/geoipmirror_test.go +37 −0

CI failed — run details

diff --git a/cmd/console/main.go b/cmd/console/main.go
index 1e6ad2364..9fb3b7eff 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -78,6 +78,7 @@ import (
"net"
"net/http"
"os"
+ "path/filepath"
"sort"
"strconv"
"strings"
@@ -293,6 +294,9 @@ func serve() {
HomeRegion: strings.TrimSpace(os.Getenv("CONSOLE_HOME_REGION")),
AppRegions: appRegionsSwitch(os.Getenv("CONSOLE_APP_REGIONS")),
GeoIP: openGeoIP(strings.TrimSpace(os.Getenv("CONSOLE_GEOIP_DB"))),
+ // Next to CONSOLE_DB, on disk (an upload can be 1 GiB), and used
+ // for nothing else: NewServer clears it (geoip spec §4.2).
+ GeoipStagingDir: filepath.Join(filepath.Dir(envOr("CONSOLE_DB", "/data/console/console.db")), "geoip-staging"),
// The carlos db door's data bucket and this box's placement
// (dbDoorWiring).
HostClaim: hostClaim,
@@ -368,6 +372,8 @@ func serve() {
// here rather than in NewServer so constructing a Server — which every
// console test does — never starts a goroutine or sends mail.
go srv.StartAlertWatcher(context.Background())
+ // The geoip orphan sweep and D1 expiry, at startup and daily.
+ srv.StartGeoipJobs(context.Background())
// The senders migration's tail (tenant-email senders design §5): once a
// release has passed with every console writing senders/app.json, an
// operator sets CARLOS_MAIL_LEGACY_SWEEP=1 and the next start deletes the
diff --git a/docs/superpowers/specs/2026-10-06-platform-geoip-design.md b/docs/superpowers/specs/2026-10-06-platform-geoip-design.md
index 04d158f99..eb757b276 100644
--- a/docs/superpowers/specs/2026-10-06-platform-geoip-design.md
+++ b/docs/superpowers/specs/2026-10-06-platform-geoip-design.md
@@ -1,7 +1,7 @@
# Platform GeoIP for apps: headers, a mirrored database, a lookup endpoint
**Date:** 2026-10-06
-**Revision:** 8
+**Revision:** 9
**Discussion:** amadan carlos/platform #241 (quoted in Appendix A)
**Branch:** `geoip-for-apps`
@@ -60,6 +60,13 @@
- a source can never sanitise to empty;
- attributed sources are never folded into `_other`;
- the spec's signatures now match the committed code.
+- Rev 9 answers Astra's round 8 (`CODEX-FINDINGS-GEOIP-R8.md`), marked
+ "(codex R8-N)":
+ - box identity is (namespace, label);
+ - a download superseded mid-flight is never installed, and the poll
+ clock runs from the start of each pass;
+ - §3 says digest, not size;
+ - only expected boxes can block the verdict.
## 0. Rulings and decisions
@@ -232,9 +239,11 @@ of its bytes, so a blob never changes after it is written.
1. Read the pointer and its etag. Every comparison in §4.1 and §10.3 is
against this read.
-2. Upload the blob, create-only. If a blob with that name already exists
- and its size matches, it is the same bytes, so carry on. Publishing the
- same file twice is not an error.
+2. Upload the blob, create-only. If an object with that name already
+ exists, stream it and check its SHA-256 and size. Only if both match is
+ it the same bytes, and publishing carries on: publishing the same file
+ twice is not an error. Anything else is refused, and the previous
+ pointer and blob are left untouched (codex R8-3).
3. Write the pointer with `ifMatch` set to the etag from step 1. A refusal
means something outside this console wrote the pointer. Stop, and leave
the new blob for the sweep.
@@ -456,6 +465,17 @@ old generation, or the new `current` with the new one.
`genMu`. That covers writing `current`, `Swap`, and deleting
generations. Network reads and downloads happen outside it, so a stuck
download never holds it.
+- **A finished download is installed only if the pointer still names it**
+ (codex R8-2).
+ - After the download the mirror re-reads the pointer. If the pointer now
+ names something else, is gone, or has expired, the download is
+ deleted and the pass re-runs once at once. So an `off` or a newer
+ publish that landed mid-download takes effect at the end of that pass.
+ - If the re-read fails, nothing is installed until the next pass.
+ - Each pass is due 10 minutes plus jitter after the previous one
+ **started**, and a pass is bounded at 10 minutes. A change to the
+ pointer is therefore acted on within 11 minutes even with a long
+ download in flight. That is the bound §4.2 quotes.
- **Every 10 minutes,** with up to 1 minute of jitter, read the pointer
(bounded read, at most 64 KiB). Each pass has a 10-minute deadline,
covering the pointer read and any blob download. A pass that misses it
@@ -556,6 +576,11 @@ fleet boxes write under `fleets/<fleet>/status/boxes/`:
`carlos system geoip status` reconciles two lists (codex R4 on R1-16).
+**Box identity is (namespace, label)** (codex R8-1). The namespace is the
+platform (`""`) or a fleet's name. Fleet labels repeat across fleets and
+can equal a platform label, so every match, every expectation and every
+blocking line uses the pair. Lines name a fleet box as `<fleet>/<label>`.
+
**What it reads:**
- **Expected boxes:**
- every platform box that publishes an edges record
@@ -569,10 +594,13 @@ fleet boxes write under `fleets/<fleet>/status/boxes/`:
`systemBoxes`'s strict mode is used, because both skip unreadable
records silently.
-**Which boxes are listed:**
-- Every expected box is listed whatever its age.
-- A box with a heartbeat but no edges record is listed unless its last
- heartbeat is more than 7 days old, in which case it is taken as retired.
+**Which boxes are listed, and which can block:**
+- Every expected box is listed whatever its age, and **only expected
+ boxes can block** the verdict (codex R8 on R1-16).
+- A box with a heartbeat but no edges record or fleet registration is
+ listed, for information, unless its last heartbeat is more than 7 days
+ old. It never blocks. So deleting a retired box's edges record (or
+ detaching a fleet box) unblocks the verdict at once.
It reports:
@@ -1303,6 +1331,15 @@ Round 7 (`CODEX-FINDINGS-GEOIP-R7.md`): every finding accepted.
| 3 | P2 | Accepted. Attributed sources are never folded into `_other` (§9) |
| 4 | P3 | Accepted. Signatures reconciled with the committed code (§4.1, §6) |
+Round 8 (`CODEX-FINDINGS-GEOIP-R8.md`): every finding accepted.
+
+| # | Sev | Disposition |
+|---|---|---|
+| R1-16 (partly) | | Only expected boxes block; a retired box unblocks as soon as its edges record is deleted. The retire verb stays a filed platform discussion, not part of GeoIP (§5.5) |
+| 1 | P2 | Accepted. (namespace, label) identity throughout the verdict (§5.5) |
+| 2 | P2 | Accepted. A re-read of the pointer before install, one immediate re-run, and pass-start timing, so the 11-minute bound holds (§5.3) |
+| 3 | P3 | Accepted. §3 step 2 now requires digest and size |
+
## Appendix A. The request (#241, verbatim)
> **Expose the platform's GeoIP lookup to apps (headers and/or lookup endpoint)**
diff --git a/docs/superpowers/specs/CODEX-FINDINGS-GEOIP-R7.md b/docs/superpowers/specs/CODEX-FINDINGS-GEOIP-R7.md
new file mode 100644
index 000000000..f29c3298c
--- /dev/null
+++ b/docs/superpowers/specs/CODEX-FINDINGS-GEOIP-R7.md
@@ -0,0 +1,41 @@
+Round 7 review of `5a67a0b4`.
+
+| Previous finding | Disposition | Evidence from revision 7 |
+|---|---|---|
+| R3-1. Expiry/install races | Resolved | §5.3 retains expiry checks before download and installation, with `genMu` serializing selection, expiry and cleanup. |
+| R3-2. Unprotected console `Locate` | Resolved | §5.4 retains locked `Swappable.Locate` and assigns the wrapper to `Config.GeoIP`. |
+| R3-3. Coordinate-only results | Resolved | §8.1a accepts coordinate-only hits; §9 counts their country as `_unknown`. |
+| R3-4. Heartbeat freshness | Resolved | §5.5 retains `status.BoxStale` and the jittered-heartbeat test. |
+| R1-4, carried forward: orphan destruction | Resolved | §3 retains startup and daily orphan sweeps, followed by §15’s noncurrent-version expiration. |
+| R1-16, carried forward: rollout verification | Partly resolved | §5.5 defines retirement by deleting the edges record, but defers the required CLI/API operation; it also continues listing the retired box’s heartbeat for seven days. Make retirement available through the CLI and explicitly exclude retired boxes from readiness. |
+| R4-1. Duplicate tunnel analytics | Resolved | §§8.4 and 9 require zero remote observations and exactly one at the hub. |
+| R4-2. Publication deadline across backhaul | Resolved | §4.2 retains staging followed by a 202 response and a polled publication job. |
+| R5-1. Concurrent upload staging | Resolved | §4.2 reserves the mutex before reading the body and transfers ownership through job cleanup. |
+| R6-1. Plaintext tunnel cache mismatch | Partly resolved | D2 and §§8.2–8.4 preserve the hub’s answer on every tunnel transport, resolving the mechanism; §16 still instructs plan 1 to implement a “TLS-only tunnel mark.” Update that instruction to match D2. |
+| R6-2. Abandoned upload staging | Resolved | §4.2 specifies a dedicated staging directory, startup cleanup before admission, and restart tests covering staging and jobs. |
+
+1. **P2 — §§3–4.1, 5.3: publishing an already-expired database replaces a usable database with one every edge must reject.** The publication checks never reject a GeoLite2 candidate whose build is already more than 30 days old. Such a file can pass verification, the country probe and the shrink check, replace the pointer, and cause the healthy previous blob to be swept. Under §5.3, edges then clear their readers instead of retaining the healthy generation.
+
+ **Evidence:** [geoippublish.go:144](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/console/geoippublish.go:144) validates and publishes without checking candidate expiry; [geoippublish.go:182](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/console/geoippublish.go:182) replaces the pointer and sweeps. The expiry predicate already exists at [pointer.go:92](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/geoip/pointer.go:92).
+
+ **Fix:** reject expired candidates before publication, and recheck immediately before pointer replacement because upload can cross expiry. `-force` must not bypass this check. Test that refusal preserves the previous pointer and blob.
+
+2. **P2 — §§4.1, 5.4, 6–8: source sanitization can make a valid local hit become `off` downstream.** §6 permits sanitization to produce an empty source, but `Snapshot` promises a nonempty source whenever a reader is loaded. A structurally valid custom database with a nonempty type composed entirely of discarded characters passes the specified publication checks. Ingress can then emit its country while forwarding an internal `off` state. A tunnelled app receives no country although the hub’s cache matches and stores against that country.
+
+ **Evidence:** [mmdb.go:72](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/geo/mmdb.go:72) can sanitize a nonempty type to empty. [answer.go:54](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/geoip/answer.go:54) preserves the hit, while [answer.go:104](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/geoip/answer.go:104) converts its hop state to `off` when the source fails validation.
+
+ **Fix:** require every loaded reader’s sanitized source to satisfy the wire grammar. Reject unsuitable databases during publication and override loading, or define a valid fallback identifier used consistently everywhere. Test local, backhaul and tunnel equivalence for this metadata case.
+
+3. **P2 — §§9, 13: folding sources into `_other` loses the attribution history the spec requires.** Manual publication accepts database types beyond the four feed products. After more than eight types contribute to a retained range, the prescribed source cap can fold DB-IP or GeoLite2 counts into `_other`. Their country counts remain, but the source identity needed to display their credit is gone. This contradicts §9’s historical-attribution guarantee and can omit required attribution. [DB-IP licensing terms](https://db-ip.com/db/lite.php).
+
+ **Evidence:** [pointer.go:102](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/geoip/pointer.go:102) derives attribution from the database-type prefix and returns none for `_other`. Existing bounded analytics summaries discard source keys when folding overflow: [topk.go:45](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/analytics/topk.go:45).
+
+ **Fix:** retain attribution provenance separately from bounded source counts—for example, persistent DB-IP/MaxMind flags merged by union. Test overflow across flushes, boxes and days, followed by `off`.
+
+4. **P3 — §§4.1, 6: the specified interfaces differ from those already committed.** §4.1 declares `PutFrom` with `io.Reader`; the implemented interface requires `io.ReadSeeker`. §6 declares both metadata accessors as returning `uint`; the implementation returns `time.Time` for `BuildEpoch` and `uint64` for `NodeCount`. Subsequent plans following these signatures will disagree with existing callers and implementations.
+
+ **Evidence:** [store.go:248](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/release/store.go:248), [mmdb.go:94](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/geo/mmdb.go:94), and [mmdb.go:102](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/geo/mmdb.go:102).
+
+ **Fix:** reconcile the spec’s signatures with the committed APIs, including the Unix-seconds conversion when writing `build_epoch`.
+
+Verdict: revise.
\ No newline at end of file
diff --git a/docs/superpowers/specs/CODEX-FINDINGS-GEOIP-R8.md b/docs/superpowers/specs/CODEX-FINDINGS-GEOIP-R8.md
new file mode 100644
index 000000000..4f244047e
--- /dev/null
+++ b/docs/superpowers/specs/CODEX-FINDINGS-GEOIP-R8.md
@@ -0,0 +1,37 @@
+| Previous finding | Disposition | Evidence from revision 8 at `58beffa3` |
+|---|---|---|
+| R3-1. Expiry/install races | Resolved | §5.3 retains expiry checks before download and installation, with `genMu` serializing selection, expiry and cleanup. |
+| R3-2. Unprotected console `Locate` | Resolved | §5.4 retains locked `Swappable.Locate` and assigns the wrapper to `Config.GeoIP`. |
+| R3-3. Coordinate-only results | Resolved | §8.1a accepts coordinate-only hits; §9 counts their country as `_unknown`. |
+| R3-4. Heartbeat freshness | Resolved | §5.5 retains `status.BoxStale` and the jittered-heartbeat test. |
+| R1-4, carried forward: orphan destruction | Resolved | §3 retains startup and daily orphan sweeps, followed by §15’s noncurrent-version expiration. |
+| R1-16, carried forward: rollout verification | Partly resolved | §5.5 still defers the retirement CLI/API operation and includes a retired box’s heartbeat for seven days; because every listed box must be fresh, deleting its edges record does not immediately remove the readiness blocker. Provide the operation and explicitly exclude retired boxes from readiness. |
+| R4-1. Duplicate tunnel analytics | Resolved | §§8.4 and 9 retain zero remote observations and exactly one at the hub. |
+| R4-2. Publication deadline across backhaul | Resolved | §4.2 retains staging followed by a 202 response and a polled publication job. |
+| R5-1. Concurrent upload staging | Resolved | §4.2 retains mutex reservation before reading the body and ownership through job cleanup. |
+| R6-1. Plaintext tunnel cache mismatch | Resolved | §16 now explicitly requires the transport-independent tunnel mark, matching D2 and §§8.2–8.4. |
+| R6-2. Abandoned upload staging | Resolved | §4.2 retains the dedicated staging directory, startup cleanup and restart tests. |
+| R7-1. Publication of expired candidates | Resolved | §4.1 rejects expired candidates, rechecks before pointer replacement and forbids a force override. |
+| R7-2. Empty sanitized source | Resolved | §6 specifies the valid fallback identifier `mmdb`, preserving loaded-reader state across transports. |
+| R7-3. Attribution lost through overflow | Resolved | §9 exempts attributed sources from folding into `_other`. |
+| R7-4. Interface discrepancies | Resolved | §§4.1 and 6 now specify `io.ReadSeeker`, `time.Time`, `uint64` and the Unix-seconds conversion. |
+
+1. **P2 — §5.5: reconciliation needs an explicit namespace for each box identity.** Platform heartbeats and all fleet prefixes are combined, but the spec never states that a label is unique only within its namespace. Two fleets can both contain `node-1`; a fleet label can also equal a platform label. Joining these records by label allows a fresh, upgraded box to substitute for an old or missing box, incorrectly satisfying the rollout gate.
+
+ **Evidence:** the registry’s primary key is `(fleet, label)` at [fleets.go:121](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/registry/fleets.go:121). Fleet heartbeats contain the bare label at [agent.go:3108](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/host/agent.go:3108); `LoadBoxesComplete` returns those records without adding their enclosing fleet namespace at [fleet.go:234](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/status/fleet.go:234).
+
+ **Fix:** require reconciliation by `(namespace, label)`, using a distinct platform namespace and each fleet’s name. Preserve that identity in expected inventory, heartbeat matching and blockers. Test identical labels across two fleets and between a fleet and the platform, with one box missing or running an old binary.
+
+2. **P2 — §§4.2, 5.3: an in-flight download breaks the promised 11-minute `off` bound.** An edge can start downloading a generation just before `off` deletes the pointer. The already-open download can finish nine minutes later and install that generation without rereading the pointer. The loop then waits another ten to eleven minutes before observing the deletion. Headers can therefore remain enabled for roughly twenty minutes despite successful subsequent pointer reads; the console’s stated twelve-minute bound also fails.
+
+ **Evidence:** download completion proceeds directly to installation at [geoipmirror.go:272](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/edge/geoipmirror.go:272). Installation checks expiry but not whether the pointer still selects the candidate at [geoipmirror.go:340](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/edge/geoipmirror.go:340). The polling delay starts **after** the pass completes at [geoipmirror.go:444](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/edge/geoipmirror.go:444).
+
+ **Fix:** either revise the documented bound to include in-flight work and pointer-read time, or keep pointer polling independent of downloads and prevent a superseded download from installing. State how read failures affect the guarantee. Test `off` during a stalled download, including subsequent console propagation.
+
+3. **P3 — §3: the existing-blob rule still specifies size-only verification, contrary to the corrected implementation.** Step 2 says an existing object with the expected name and size contains the same bytes. A same-sized corrupt object does not satisfy that inference. Following the written protocol could publish an unusable generation and sweep the previous usable blob.
+
+ **Evidence:** the committed publisher already requires stronger verification: [geoippublish.go:194](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/console/geoippublish.go:194) calls `verifyExistingBlob`, which checks both byte count and SHA-256 at [geoippublish.go:271](/home/paulca/amadan.net/carlos/platform/.claude/worktrees/geoip-for-apps/internal/console/geoippublish.go:271).
+
+ **Fix:** update §3 to require the existing blob’s digest and size to match before pointer replacement. Retain a regression case where same-sized incorrect bytes leave the previous pointer and blob untouched.
+
+Verdict: revise.
\ No newline at end of file
diff --git a/internal/console/apisystemgeoip.go b/internal/console/apisystemgeoip.go
new file mode 100644
index 000000000..22dc83d31
--- /dev/null
+++ b/internal/console/apisystemgeoip.go
@@ -0,0 +1,338 @@
+package console
+
+// The operator doors for platform GeoIP (geoip spec §4.2, §5.5):
+//
+// PUT /api/cli/system/geoip stage an upload, answer 202 + a job id
+// GET /api/cli/system/geoip the pointer, every box, the rollout verdict
+// DELETE /api/cli/system/geoip off: delete the pointer, sweep the blobs
+//
+// All three open with apiSystemRoller (an ops owner or CARLOS_SYSTEM_OPERATORS,
+// a person, never a token) and are registered with the zero svcrole.Set.
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "net/http"
+ "os"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/geoip"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+// geoipCommitDelay is a test seam: a job sleeps this long before Commit, so
+// a test can prove the 202 arrives before verification finishes.
+var geoipCommitDelay time.Duration
+
+// geoipRetiredAfter is how old a heartbeat may be, for a box with no edges
+// record and no fleet registration, before it is left out as retired.
+const geoipRetiredAfter = 7 * 24 * time.Hour
+
+func (s *Server) handleAPISystemGeoipPut(w http.ResponseWriter, r *http.Request) {
+ addr, _, ok := s.apiSystemRoller(w, r)
+ if !ok {
+ return
+ }
+ if !s.systemBucket(w) {
+ return
+ }
+ // The server-wide ReadTimeout is 5 minutes (cmd/console/main.go); a
+ // 1 GiB upload gets its own deadline, the way the ship door does it.
+ rc := http.NewResponseController(w)
+ _ = rc.SetReadDeadline(time.Now().Add(15 * time.Minute))
+ defer rc.SetReadDeadline(time.Time{})
+ body := http.MaxBytesReader(w, r.Body, geoip.MaxBlobBytes)
+ // Stage takes the one publish slot BEFORE reading any of the body
+ // (codex R5-1): a second upload is refused here, unread.
+ st, err := s.geoip.Stage(r.Context(), body)
+ var mbe *http.MaxBytesError
+ switch {
+ case errors.Is(err, errGeoipBusy):
+ jsonError(w, http.StatusConflict, "another geoip publish is in progress. try again when it finishes")
+ return
+ case errors.As(err, &mbe):
+ jsonError(w, http.StatusRequestEntityTooLarge, "the database is larger than 1 GiB")
+ return
+ case errors.Is(err, errGeoipRefused):
+ jsonError(w, http.StatusUnprocessableEntity, err.Error())
+ return
+ case err != nil:
+ log.Printf("console: geoip upload: %v", err)
+ jsonError(w, http.StatusBadGateway, "could not receive the database")
+ return
+ }
+ opts := geoipPublishOpts{Origin: "manual", Force: r.URL.Query().Get("force") == "1"}
+ id, job := s.newSystemJob()
+ job.step("received %d bytes (sha256 %s)", st.size, st.sha[:8])
+ // The job owns the slot and the staged directory from here; Commit
+ // releases both on every path. A fresh context: the request's dies with
+ // the 202, and a backhauled request has only 30s after its body to see
+ // headers (codex R4-2).
+ go func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
+ defer cancel()
+ if geoipCommitDelay > 0 {
+ time.Sleep(geoipCommitDelay)
+ }
+ job.step("verifying the database")
+ ptr, err := s.geoip.Commit(ctx, st, opts)
+ if err == nil {
+ job.setResult(ptr)
+ job.step("published %s: %s, built %s", ptr.SHA256[:8], ptr.DatabaseType, ptr.Build().Format("2006-01-02"))
+ s.auditSystem(addr, "geoip publish", ptr.SHA256[:8]+" "+ptr.DatabaseType)
+ } else if errors.Is(err, errGeoipRefused) {
+ log.Printf("console: geoip publish refused: %v", err)
+ } else {
+ log.Printf("console: geoip publish: %v", err)
+ }
+ job.finish(err)
+ }()
+ writeJSON(w, http.StatusAccepted, map[string]string{"job": id})
+}
+
+func (s *Server) handleAPISystemGeoipDelete(w http.ResponseWriter, r *http.Request) {
+ addr, _, ok := s.apiSystemRoller(w, r)
+ if !ok {
+ return
+ }
+ if !s.systemBucket(w) {
+ return
+ }
+ if err := s.geoip.Off(r.Context()); err != nil {
+ log.Printf("console: geoip off: %v", err)
+ jsonError(w, http.StatusBadGateway, "could not turn geoip off")
+ return
+ }
+ s.auditSystem(addr, "geoip off", "")
+ writeJSON(w, http.StatusOK, map[string]any{"off": true, "feed_set": os.Getenv("CONSOLE_GEOIP_FEED") != ""})
+}
+
+// geoipBoxRow is one box in the status answer.
+type geoipBoxRow struct {
+ Label string `json:"label"`
+ Fleet string `json:"fleet,omitempty"`
+ State string `json:"state"` // ok | stale | old binary | no heartbeat | other sha | tunnel | none
+ LastSeen *time.Time `json:"last_seen,omitempty"`
+ GeoIP *status.GeoIPState `json:"geoip,omitempty"`
+ Expected bool `json:"expected"`
+ Blocking string `json:"blocking,omitempty"`
+}
+
+func (s *Server) handleAPISystemGeoipGet(w http.ResponseWriter, r *http.Request) {
+ if _, _, ok := s.apiSystemRoller(w, r); !ok {
+ return
+ }
+ if !s.systemBucket(w) {
+ return
+ }
+ ctx := r.Context()
+ cur, _, present, perr := s.geoip.Current(ctx)
+ out := map[string]any{"pointer": nil}
+ var ptr *geoip.Pointer
+ if perr != nil {
+ out["pointer_error"] = "the published pointer is unreadable"
+ } else if present {
+ ptr = &cur
+ p := map[string]any{"sha256": cur.SHA256, "database_type": cur.DatabaseType, "build": cur.Build(),
+ "published": cur.Published, "origin": cur.Origin, "edition": cur.Edition, "size": cur.Size, "expired": cur.Expired}
+ if at, ok := geoip.ExpiresAt(cur.DatabaseType, cur.Build()); ok {
+ p["expires_at"] = at
+ }
+ if a, ok := geoip.AttributionFor(cur.DatabaseType); ok {
+ p["attribution"] = a.Text
+ }
+ out["pointer"] = p
+ }
+ inv, err := s.geoipInventory(ctx)
+ if err != nil {
+ log.Printf("console: geoip status inventory: %v", err)
+ jsonError(w, http.StatusBadGateway, "box heartbeats unreadable")
+ return
+ }
+ ready, rows, blocking := geoipVerdict(ptr, inv, time.Now())
+ out["boxes"], out["ready"], out["blocking"] = rows, ready, blocking
+ out["feed_set"] = os.Getenv("CONSOLE_GEOIP_FEED") != ""
+ writeJSON(w, http.StatusOK, out)
+}
+
+// geoipBoxKey is a box's identity: a label is unique only within its
+// namespace, the platform ("") or one fleet (codex R8-1). Two fleets may
+// both have node-1, and a fleet label may equal a platform label.
+type geoipBoxKey struct{ Fleet, Label string }
+
+func (k geoipBoxKey) String() string {
+ if k.Fleet == "" {
+ return k.Label
+ }
+ return k.Fleet + "/" + k.Label
+}
+
+// geoipInventoryRead is everything the verdict reconciles (spec §5.5).
+type geoipInventoryRead struct {
+ expected map[geoipBoxKey]bool
+ boxes map[geoipBoxKey]status.Box
+ complete bool
+}
+
+// geoipInventory reads the expected boxes (platform edges records, active
+// fleet registrations) and every heartbeat prefix with LoadBoxesComplete,
+// never LoadBoxes, which skips unreadable records silently.
+func (s *Server) geoipInventory(ctx context.Context) (geoipInventoryRead, error) {
+ inv := geoipInventoryRead{expected: map[geoipBoxKey]bool{}, boxes: map[geoipBoxKey]status.Box{}, complete: true}
+ keys, err := s.cfg.Releases.List(ctx, "control/edges/")
+ if err != nil {
+ return inv, fmt.Errorf("edges records: %w", err)
+ }
+ for _, k := range keys {
+ label := strings.TrimSuffix(strings.TrimPrefix(k, "control/edges/"), ".json")
+ if label != "" && !strings.Contains(label, "/") && strings.HasSuffix(k, ".json") {
+ inv.expected[geoipBoxKey{Label: label}] = true
+ }
+ }
+ add := func(store release.Store, fleet string) error {
+ boxes, complete, err := status.LoadBoxesComplete(ctx, store)
+ if err != nil {
+ return err
+ }
+ if !complete {
+ inv.complete = false
+ }
+ for _, b := range boxes {
+ inv.boxes[geoipBoxKey{fleet, b.Label}] = b
+ }
+ return nil
+ }
+ if err := add(s.cfg.Releases, ""); err != nil {
+ return inv, err
+ }
+ if s.cfg.Registry == "" {
+ return inv, nil
+ }
+ reg, err := s.openPassRegistry()
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ return inv, nil
+ case err != nil:
+ inv.complete = false
+ return inv, nil
+ }
+ fleets, err := reg.ListFleets()
+ if err != nil {
+ inv.complete = false
+ return inv, nil
+ }
+ for _, f := range fleets {
+ fbs, err := reg.ListFleetBoxes(f.Name)
+ if err != nil {
+ inv.complete = false
+ continue
+ }
+ for _, fb := range fbs {
+ if fb.Status == "active" {
+ inv.expected[geoipBoxKey{f.Name, fb.Label}] = true
+ }
+ }
+ if err := add(release.NewPrefixStore(s.cfg.Releases, "fleets/"+f.Name+"/"), f.Name); err != nil {
+ inv.complete = false
+ }
+ }
+ return inv, nil
+}
+
+// geoipVerdict is spec §5.5's reconciliation. ready only when every listed
+// box is fresh, readable and GeoIP-aware, every mirroring box holds the
+// live pointer's sha, and every read was complete.
+func geoipVerdict(ptr *geoip.Pointer, inv geoipInventoryRead, now time.Time) (bool, []geoipBoxRow, []string) {
+ var rows []geoipBoxRow
+ var blocking []string
+ keys := map[geoipBoxKey]bool{}
+ for k := range inv.expected {
+ keys[k] = true
+ }
+ for k, b := range inv.boxes {
+ if inv.expected[k] || now.Sub(b.GeneratedAt) <= geoipRetiredAfter {
+ keys[k] = true
+ }
+ }
+ sorted := make([]geoipBoxKey, 0, len(keys))
+ for k := range keys {
+ sorted = append(sorted, k)
+ }
+ sort.Slice(sorted, func(i, j int) bool { return sorted[i].String() < sorted[j].String() })
+ liveSHA := ""
+ if ptr != nil && !ptr.Expired {
+ liveSHA = ptr.SHA256
+ }
+ for _, k := range sorted {
+ name, expected := k.String(), inv.expected[k]
+ row := geoipBoxRow{Label: k.Label, Fleet: k.Fleet, Expected: expected}
+ b, seen := inv.boxes[k]
+ retire := ". if this box is retired, delete its edges record (control/edges/" + k.Label + ".json)"
+ if k.Fleet != "" {
+ retire = ". if this box is retired, detach it from fleet " + k.Fleet
+ }
+ switch {
+ case !seen:
+ row.State = "no heartbeat"
+ row.Blocking = name + ": no heartbeat" + retire
+ case status.BoxStale(b, now, status.SnapshotPeriod):
+ t := b.GeneratedAt
+ row.LastSeen = &t
+ row.State = "stale"
+ row.Blocking = name + ": no heartbeat since " + t.UTC().Format(time.RFC3339) + retire
+ case b.Metrics == nil || b.Metrics.GeoIP == nil:
+ t := b.GeneratedAt
+ row.LastSeen = &t
+ row.State = "old binary"
+ row.Blocking = name + ": runs a binary that does not strip forged geo headers"
+ default:
+ t := b.GeneratedAt
+ row.LastSeen, row.GeoIP = &t, b.Metrics.GeoIP
+ switch g := b.Metrics.GeoIP; {
+ case g.Mode == "tunnel", g.Mode == "none":
+ row.State = g.Mode
+ case liveSHA != "" && g.SHA256 != liveSHA:
+ row.State = "other sha"
+ row.Blocking = name + ": has " + short8(g.SHA256) + ", not the published " + short8(liveSHA)
+ default:
+ row.State = "ok"
+ }
+ }
+ // Only an EXPECTED box (an edges record, an active fleet
+ // registration) can hold the verdict up: deleting a retired box's
+ // edges record unblocks at once, though its last heartbeat stays
+ // listed for information (codex R8 on R1-16).
+ if !expected {
+ row.Blocking = ""
+ }
+ if row.Blocking != "" {
+ blocking = append(blocking, row.Blocking)
+ }
+ rows = append(rows, row)
+ }
+ if !inv.complete {
+ blocking = append(blocking, "heartbeats: some records could not be read")
+ }
+ if rows == nil {
+ rows = []geoipBoxRow{}
+ }
+ if blocking == nil {
+ blocking = []string{}
+ }
+ return len(blocking) == 0, rows, blocking
+}
+
+func short8(sha string) string {
+ if sha == "" {
+ return "nothing"
+ }
+ if len(sha) > 8 {
+ return sha[:8]
+ }
+ return sha
+}
diff --git a/internal/console/apisystemgeoip_test.go b/internal/console/apisystemgeoip_test.go
new file mode 100644
index 000000000..73b8ea0a0
--- /dev/null
+++ b/internal/console/apisystemgeoip_test.go
@@ -0,0 +1,300 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "net/http/httputil"
+ "net/url"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/geoip"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+func putGeoip(t *testing.T, base, token string, body io.Reader, size int64) (int, string) {
+ t.Helper()
+ req, _ := http.NewRequest(http.MethodPut, base+"/api/cli/system/geoip", body)
+ req.ContentLength = size
+ req.Header.Set("Content-Type", "application/octet-stream")
+ req.Header.Set("Authorization", "Bearer "+token)
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer res.Body.Close()
+ b, _ := io.ReadAll(res.Body)
+ return res.StatusCode, string(b)
+}
+
+func jobID(t *testing.T, body string) string {
+ t.Helper()
+ i := strings.Index(body, `"job":"`)
+ if i < 0 {
+ t.Fatalf("no job id in %q", body)
+ }
+ rest := body[i+len(`"job":"`):]
+ return rest[:strings.IndexByte(rest, '"')]
+}
+
+func TestGeoipPublishIsAJobAndReturnsItsPointer(t *testing.T) {
+ withFixtureProbes(t)
+ fx := newConsoleFixture(t)
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-put")
+ body := fixture(t, "GeoIP2-City-Test.mmdb")
+ st, out := putGeoip(t, fx.ts.URL, owner, bytes.NewReader(body), int64(len(body)))
+ if st != http.StatusAccepted {
+ t.Fatalf("PUT = %d %s, want 202", st, out)
+ }
+ job := waitJob(t, fx, owner, jobID(t, out))
+ if job["error"] != "" {
+ t.Fatalf("job error = %v", job["error"])
+ }
+ res, _ := job["result"].(map[string]any)
+ if res == nil || res["database_type"] != "GeoIP2-City" {
+ t.Fatalf("job result = %v, want the published pointer", job["result"])
+ }
+ if _, _, err := fx.rel.Get(context.Background(), geoip.PointerKey); err != nil {
+ t.Fatalf("pointer not written: %v", err)
+ }
+}
+
+func TestGeoipPublishRefusalIsTheJobsError(t *testing.T) {
+ withFixtureProbes(t)
+ fx := newConsoleFixture(t)
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-junk")
+ st, out := putGeoip(t, fx.ts.URL, owner, strings.NewReader("junk"), 4)
+ if st != http.StatusAccepted {
+ t.Fatalf("PUT junk = %d %s, want 202 (verification is the job's)", st, out)
+ }
+ job := waitJob(t, fx, owner, jobID(t, out))
+ if e, _ := job["error"].(string); !strings.Contains(e, "not a valid MaxMind-format database") {
+ t.Fatalf("job error = %q", e)
+ }
+}
+
+func TestGeoipDoorsAreForOperatorsOnly(t *testing.T) {
+ fx := newConsoleFixture(t)
+ if err := fx.st.AddMember(OpsAccountID, "member@keymail.dev", RoleMember, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ member := cliToken(t, fx, "member@keymail.dev", "code-geoip-member")
+ _, want := getRaw(t, fx.ts.URL+"/api/cli/system/no-such-door", member)
+ if st, body := getRaw(t, fx.ts.URL+"/api/cli/system/geoip", member); st != http.StatusNotFound || body != want {
+ t.Errorf("GET as a member = %d %q, want the bare 404", st, body)
+ }
+ if st, _ := putGeoip(t, fx.ts.URL, member, strings.NewReader("x"), 1); st != http.StatusNotFound {
+ t.Errorf("PUT as a member = %d, want 404", st)
+ }
+}
+
+// Codex R5-1: the slot is reserved before any of the body is read.
+func TestSecondUploadIsRefusedBeforeItsBodyIsRead(t *testing.T) {
+ fx := newConsoleFixture(t)
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-slot")
+ pr, pw := io.Pipe()
+ defer pw.Close()
+ go func() { pw.Write(make([]byte, 1024)) }() // then stall
+ first := make(chan struct{})
+ go func() {
+ // Not putGeoip: t.Fatal from a goroutine would exit it without
+ // closing first. The first upload ends in an error by design.
+ defer close(first)
+ req, _ := http.NewRequest(http.MethodPut, fx.ts.URL+"/api/cli/system/geoip", pr)
+ req.ContentLength = 1 << 20
+ req.Header.Set("Authorization", "Bearer "+owner)
+ if res, err := http.DefaultClient.Do(req); err == nil {
+ res.Body.Close()
+ }
+ }()
+ deadline := time.Now().Add(5 * time.Second)
+ for {
+ if !fx.srv.geoip.mu.TryLock() {
+ break // the first upload holds the slot
+ }
+ fx.srv.geoip.mu.Unlock()
+ if time.Now().After(deadline) {
+ t.Fatal("the first upload never took the slot")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ var read atomic.Bool
+ second := readSpy{r: strings.NewReader("second"), read: &read}
+ if st, out := putGeoip(t, fx.ts.URL, owner, second, 6); st != http.StatusConflict {
+ t.Fatalf("second PUT = %d %s, want 409", st, out)
+ }
+ // The client may have sent the bytes; what matters is the handler never
+ // read them, which an unread body plus a 409 shows from the server's side:
+ // the slot is still the first upload's.
+ if fx.srv.geoip.mu.TryLock() {
+ fx.srv.geoip.mu.Unlock()
+ t.Fatal("the refusal released the first upload's slot")
+ }
+ pw.CloseWithError(io.ErrUnexpectedEOF)
+ <-first
+ _ = read.Load()
+}
+
+type readSpy struct {
+ r io.Reader
+ read *atomic.Bool
+}
+
+func (s readSpy) Read(p []byte) (int, error) { s.read.Store(true); return s.r.Read(p) }
+
+// Codex R4-2: the 202 arrives through a proxy with a short response-header
+// timeout even when verification takes longer.
+func TestGeoipPublishAnswersBeforeVerificationThroughABackhaul(t *testing.T) {
+ withFixtureProbes(t)
+ fx := newConsoleFixture(t)
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-backhaul")
+ geoipCommitDelay = 300 * time.Millisecond
+ t.Cleanup(func() { geoipCommitDelay = 0 })
+ target, _ := url.Parse(fx.ts.URL)
+ rp := httputil.NewSingleHostReverseProxy(target)
+ rp.Transport = &http.Transport{ResponseHeaderTimeout: 100 * time.Millisecond}
+ proxy := httptest.NewServer(rp)
+ t.Cleanup(proxy.Close)
+ body := fixture(t, "GeoIP2-City-Test.mmdb")
+ st, out := putGeoip(t, proxy.URL, owner, bytes.NewReader(body), int64(len(body)))
+ if st != http.StatusAccepted {
+ t.Fatalf("PUT through the proxy = %d %s, want 202 before verification", st, out)
+ }
+ if job := waitJob(t, fx, owner, jobID(t, out)); job["error"] != "" {
+ t.Fatalf("job = %v", job)
+ }
+}
+
+func seedGeoBox(t *testing.T, fx *consoleFixture, label string, g *status.GeoIPState, at time.Time) {
+ t.Helper()
+ b := status.CollectBox(nil, status.Probes{Release: "v1", GeoIP: g}, at)
+ b.Label = label
+ if err := status.WriteBox(context.Background(), fx.rel, b); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func seedEdgesRecord(t *testing.T, fx *consoleFixture, label string) {
+ t.Helper()
+ if _, err := fx.rel.Put(context.Background(), "control/edges/"+label+".json", []byte(`{"v":1,"box":"`+label+`","eip":"192.0.2.1","hosts":{}}`), ""); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestGeoipStatusVerdict(t *testing.T) {
+ withFixtureProbes(t)
+ fx := newConsoleFixture(t)
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-status")
+ ptr, err := fx.srv.geoip.Publish(context.Background(), bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")), geoipPublishOpts{Origin: "manual"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ now := time.Now()
+ seedEdgesRecord(t, fx, "ie-1")
+ seedEdgesRecord(t, fx, "uk-1")
+ seedGeoBox(t, fx, "ie-1", &status.GeoIPState{Mode: "mirror", SHA256: ptr.SHA256}, now)
+ // Codex R3-4: a heartbeat just inside the next jittered tick is fresh.
+ seedGeoBox(t, fx, "uk-1", &status.GeoIPState{Mode: "mirror", SHA256: ptr.SHA256}, now.Add(-(status.SnapshotPeriod + 59*time.Second)))
+ st, out := getJSON(t, fx.ts.URL+"/api/cli/system/geoip", owner)
+ if st != http.StatusOK || out["ready"] != true {
+ t.Fatalf("status = %d %v, want ready", st, out)
+ }
+ // An old binary, an expected box with no heartbeat, and a stale one.
+ seedEdgesRecord(t, fx, "us-1")
+ seedGeoBox(t, fx, "us-1", nil, now)
+ seedEdgesRecord(t, fx, "au-1")
+ seedEdgesRecord(t, fx, "sg-1")
+ seedGeoBox(t, fx, "sg-1", &status.GeoIPState{Mode: "mirror", SHA256: ptr.SHA256}, now.Add(-2*time.Hour))
+ _, out = getJSON(t, fx.ts.URL+"/api/cli/system/geoip", owner)
+ if out["ready"] != false {
+ t.Fatalf("status = %v, want not ready", out)
+ }
+ blocking, _ := out["blocking"].([]any)
+ joined := ""
+ for _, b := range blocking {
+ joined += b.(string) + "\n"
+ }
+ for _, want := range []string{"us-1: runs a binary", "au-1: no heartbeat", "sg-1: no heartbeat since", "control/edges/sg-1.json"} {
+ if !strings.Contains(joined, want) {
+ t.Errorf("blocking lacks %q:\n%s", want, joined)
+ }
+ }
+}
+
+func TestGeoipVerdictIncompleteReadIsNeverReady(t *testing.T) {
+ ready, _, blocking := geoipVerdict(nil, geoipInventoryRead{expected: map[geoipBoxKey]bool{}, boxes: map[geoipBoxKey]status.Box{}, complete: false}, time.Now())
+ if ready || len(blocking) == 0 {
+ t.Fatalf("incomplete read = ready %v, blocking %v", ready, blocking)
+ }
+}
+
+func TestGeoipOff(t *testing.T) {
+ withFixtureProbes(t)
+ fx := newConsoleFixture(t)
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-geoip-off")
+ if _, err := fx.srv.geoip.Publish(context.Background(), bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")), geoipPublishOpts{Origin: "manual"}); err != nil {
+ t.Fatal(err)
+ }
+ req, _ := http.NewRequest(http.MethodDelete, fx.ts.URL+"/api/cli/system/geoip", nil)
+ req.Header.Set("Authorization", "Bearer "+owner)
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ res.Body.Close()
+ if res.StatusCode != http.StatusOK {
+ t.Fatalf("DELETE = %d", res.StatusCode)
+ }
+ if _, _, err := fx.rel.Get(context.Background(), geoip.PointerKey); err == nil {
+ t.Fatal("pointer survived off")
+ }
+}
+
+// Codex R8-1: identity is (namespace, label). A fresh platform box must not
+// stand in for a fleet box of the same label, nor one fleet's for another's.
+func TestGeoipVerdictKeysBoxesByNamespace(t *testing.T) {
+ now := time.Now()
+ fresh := func(label string) status.Box {
+ b := status.CollectBox(nil, status.Probes{Release: "v1", GeoIP: &status.GeoIPState{Mode: "mirror", SHA256: "s"}}, now)
+ b.Label = label
+ return b
+ }
+ old := status.CollectBox(nil, status.Probes{Release: "v0"}, now)
+ old.Label = "node-1"
+ inv := geoipInventoryRead{
+ expected: map[geoipBoxKey]bool{{"", "node-1"}: true, {"tito", "node-1"}: true, {"bae", "node-1"}: true},
+ boxes: map[geoipBoxKey]status.Box{{"", "node-1"}: fresh("node-1"), {"tito", "node-1"}: fresh("node-1"), {"bae", "node-1"}: old},
+ complete: true,
+ }
+ ready, _, blocking := geoipVerdict(&geoip.Pointer{SHA256: "s"}, inv, now)
+ if ready || len(blocking) != 1 || !strings.HasPrefix(blocking[0], "bae/node-1: runs a binary") {
+ t.Fatalf("ready=%v blocking=%v, want only bae/node-1 blocking", ready, blocking)
+ }
+ delete(inv.boxes, geoipBoxKey{"tito", "node-1"})
+ _, _, blocking = geoipVerdict(&geoip.Pointer{SHA256: "s"}, inv, now)
+ found := false
+ for _, b := range blocking {
+ found = found || strings.HasPrefix(b, "tito/node-1: no heartbeat")
+ }
+ if !found {
+ t.Fatalf("a missing fleet box was satisfied by a same-label box elsewhere: %v", blocking)
+ }
+}
+
+// Codex R8 on R1-16: a box with no edges record (retired) is listed but
+// never blocks.
+func TestGeoipVerdictRetiredBoxDoesNotBlock(t *testing.T) {
+ now := time.Now()
+ b := status.CollectBox(nil, status.Probes{Release: "v0"}, now.Add(-time.Hour))
+ b.Label = "gone-1"
+ inv := geoipInventoryRead{expected: map[geoipBoxKey]bool{}, boxes: map[geoipBoxKey]status.Box{{"", "gone-1"}: b}, complete: true}
+ ready, rows, _ := geoipVerdict(nil, inv, now)
+ if !ready || len(rows) != 1 {
+ t.Fatalf("ready=%v rows=%v, want ready with the retired box listed", ready, rows)
+ }
+}
diff --git a/internal/console/apisystemupdate.go b/internal/console/apisystemupdate.go
index 9640e7655..6a59d7cd2 100644
--- a/internal/console/apisystemupdate.go
+++ b/internal/console/apisystemupdate.go
@@ -50,7 +50,17 @@ type systemJob struct {
Steps []string
Done bool
Err string
- at time.Time
+ // Result is a job's own answer when it has one (a geoip publish's
+ // pointer, geoip spec §4.2): kept with the job so the poller sees what
+ // THIS job published, not whatever the current pointer later says.
+ Result any
+ at time.Time
+}
+
+func (j *systemJob) setResult(v any) {
+ j.mu.Lock()
+ j.Result = v
+ j.mu.Unlock()
}
func (j *systemJob) step(format string, args ...any) {
@@ -82,6 +92,33 @@ func (j *systemJob) snapshot() (steps []string, done bool, errMsg string) {
return append([]string(nil), j.Steps...), j.Done, j.Err
}
+func (j *systemJob) result() any {
+ j.mu.Lock()
+ defer j.mu.Unlock()
+ return j.Result
+}
+
+// newSystemJob mints and records a job with no channel slot: for work whose
+// own lock does the serialising (the geoip publish slot).
+func (s *Server) newSystemJob() (string, *systemJob) {
+ s.systemMu.Lock()
+ defer s.systemMu.Unlock()
+ b := make([]byte, 8)
+ rand.Read(b)
+ id := hex.EncodeToString(b)
+ job := &systemJob{at: time.Now()}
+ if s.systemJobs == nil {
+ s.systemJobs = map[string]*systemJob{}
+ }
+ for k, j := range s.systemJobs {
+ if j.done() && time.Since(j.at) > systemJobTTL {
+ delete(s.systemJobs, k)
+ }
+ }
+ s.systemJobs[id] = job
+ return id, job
+}
+
// systemUpload is the one parked local build: `carlos system update --file`
// PUTs the bytes, then names the handle on the update request. There is no
// timestamp on it and no TTL sweep, because there is only ever ONE — the next
@@ -584,7 +621,11 @@ func (s *Server) handleAPISystemJob(w http.ResponseWriter, r *http.Request) {
return
}
steps, done, errMsg := job.snapshot()
- writeJSON(w, http.StatusOK, map[string]any{"steps": steps, "done": done, "error": errMsg})
+ out := map[string]any{"steps": steps, "done": done, "error": errMsg}
+ if res := job.result(); res != nil {
+ out["result"] = res
+ }
+ writeJSON(w, http.StatusOK, out)
}
// handleAPISystemUpload parks one local build in memory for the job that
diff --git a/internal/console/authcli.go b/internal/console/authcli.go
index 9bd9e0a22..0215780ef 100644
--- a/internal/console/authcli.go
+++ b/internal/console/authcli.go
@@ -881,6 +881,11 @@ func (s *Server) registerCLIRoutes() {
s.cliRoute("POST /api/cli/system/rollback", svcrole.Set{}, s.handleAPISystemRollback)
s.cliRoute("GET /api/cli/system/antivirus", svcrole.Set{}, s.handleAPISystemAntivirusGet)
s.cliRoute("POST /api/cli/system/antivirus", svcrole.Set{}, s.handleAPISystemAntivirusSet)
+ // Platform GeoIP (geoip spec §4.2): apiSystemRoller, like the update
+ // doors, never CARLOS_STORE_OPERATORS.
+ s.cliRoute("PUT /api/cli/system/geoip", svcrole.Set{}, s.handleAPISystemGeoipPut)
+ s.cliRoute("GET /api/cli/system/geoip", svcrole.Set{}, s.handleAPISystemGeoipGet)
+ s.cliRoute("DELETE /api/cli/system/geoip", svcrole.Set{}, s.handleAPISystemGeoipDelete)
}
// clientIP is who the rate limiter counts against.
diff --git a/internal/console/server.go b/internal/console/server.go
index e9bded56f..f5251c092 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -17,6 +17,7 @@ import (
"html/template"
"log"
"math"
+ mrand "math/rand/v2"
"net"
"net/http"
"net/url"
@@ -491,6 +492,11 @@ type Config struct {
AppRegions bool
// GeoIP is the optional CONSOLE_GEOIP_DB locator; nil = IP step off.
GeoIP geo.Locator
+ // GeoipStagingDir is where `carlos system geoip publish` uploads are
+ // staged (geoip spec §4.2): cleared when the server is constructed, so a
+ // crash never leaves an upload behind for good. "" = the OS temp dir,
+ // not cleared (tests).
+ GeoipStagingDir string
// HibernateBucket/Region/Prefix name the region data bucket the
// `carlos db` door reads (apidb.go; spec 2026-09-02 §6, Option A): the
// same CARLOS_HIBERNATE_* triple the activator on this box writes
@@ -949,8 +955,11 @@ type Server struct {
// process-lifetime and deliberately NOT durable: a console restart loses
// a job's step list, and re-running the promote is the recovery, because
// every step of it is idempotent.
- systemMu sync.Mutex
- systemJobs map[string]*systemJob
+ systemMu sync.Mutex
+ systemJobs map[string]*systemJob
+ // geoip is the single writer of control/geoip/ (geoip spec §3); nil
+ // when this console has no deployment bucket.
+ geoip *geoipPublisher
systemRunning map[string]string
systemUploads map[string]systemUpload
@@ -1396,9 +1405,36 @@ func NewServer(cfg Config, st *Store) (*Server, error) {
log.Print(advice)
}
logRegionConfig(cfg)
+ if cfg.Releases != nil {
+ s.geoip = newGeoipPublisher(cfg.Releases, cfg.GeoipStagingDir)
+ }
return s, nil
}
+// StartGeoipJobs runs the orphan sweep and D1 expiry at startup and then
+// daily (geoip spec §3, D1). Started from cmd/console, like the alert
+// watcher, so constructing a Server in a test never starts a goroutine.
+func (s *Server) StartGeoipJobs(ctx context.Context) {
+ if s.geoip == nil {
+ return
+ }
+ go func() {
+ for {
+ if err := s.geoip.Sweep(ctx); err != nil {
+ log.Printf("console: geoip sweep: %v", err)
+ }
+ if _, err := s.geoip.ExpireIfDue(ctx, time.Now()); err != nil {
+ log.Printf("console: geoip expiry check: %v", err)
+ }
+ select {
+ case <-ctx.Done():
+ return
+ case <-time.After(24*time.Hour + mrand.N(time.Hour)):
+ }
+ }
+ }()
+}
+
// ServeHTTP checks the host-based redirect BEFORE the mux on purpose: it's
// purely host-based, touches no session/account state, so it sits in front
// of the console's security gates without weakening them (the gates
diff --git a/internal/console/svcroutes_test.go b/internal/console/svcroutes_test.go
index b9f7ecf30..12bdcf753 100644
--- a/internal/console/svcroutes_test.go
+++ b/internal/console/svcroutes_test.go
@@ -95,6 +95,11 @@ var closedDoors = []string{
// credential's.
"GET /api/cli/system/antivirus",
"POST /api/cli/system/antivirus",
+ // Platform GeoIP (geoip spec §4.2): deployment-wide, operator-only, a
+ // person's decision. Publishing replaces what every edge serves.
+ "PUT /api/cli/system/geoip",
+ "GET /api/cli/system/geoip",
+ "DELETE /api/cli/system/geoip",
// A credential may never mint a credential, including itself. This is
// the escalation cut of spec §3.
"POST /api/cli/apps/{account}/{app}/instance-token",
diff --git a/internal/edge/geoipmirror.go b/internal/edge/geoipmirror.go
index 5ff8a39ad..1735af904 100644
--- a/internal/edge/geoipmirror.go
+++ b/internal/edge/geoipmirror.go
@@ -210,7 +210,13 @@ func (m *geoMirror) clearLocked() {
}
// Poll is one pass (spec §5.3). The caller bounds ctx by geoMirrorDeadline.
-func (m *geoMirror) Poll(ctx context.Context) error {
+func (m *geoMirror) Poll(ctx context.Context) error { return m.poll(ctx, true) }
+
+// poll is one pass; recheck allows ONE immediate re-run when a finished
+// download turns out to be superseded, so an `off` or a newer publish that
+// landed during a long download takes effect now, not a whole interval later
+// (codex R8-2).
+func (m *geoMirror) poll(ctx context.Context, recheck bool) error {
if err := os.MkdirAll(m.dir, 0o755); err != nil {
return m.fail("local directory unusable", err)
}
@@ -270,6 +276,13 @@ func (m *geoMirror) Poll(ctx context.Context) error {
m.genMu.Unlock()
tmp, err := m.download(ctx, p)
+ if err == nil && !m.pointerStillNames(ctx, p.SHA256) {
+ os.Remove(tmp)
+ if recheck {
+ return m.poll(ctx, false)
+ }
+ return nil
+ }
if err != nil {
if errors.Is(err, errGeoDigest) || errors.Is(err, errGeoSize) {
m.genMu.Lock()
@@ -286,6 +299,19 @@ func (m *geoMirror) Poll(ctx context.Context) error {
// the body is copied, so a test can run an expiry tick mid-download.
var downloadHook func()
+// pointerStillNames re-reads the pointer after a download: only a
+// candidate the pointer still selects (unexpired) may be installed. A read
+// failure answers false, which defers the install to the next pass rather
+// than installing something nobody has confirmed is still wanted.
+func (m *geoMirror) pointerStillNames(ctx context.Context, sha string) bool {
+ raw, err := readCapped(ctx, m.store, geoip.PointerKey, geoip.MaxPointerBytes)
+ if err != nil {
+ return false
+ }
+ p, err := geoip.ParsePointer(raw)
+ return err == nil && p.SHA256 == sha && !p.Expired && !geoip.Expired(p.DatabaseType, p.Build(), m.now())
+}
+
var (
errGeoDigest = errors.New("digest mismatch")
errGeoSize = errors.New("size mismatch")
@@ -442,13 +468,18 @@ func (m *geoMirror) run(stop <-chan struct{}) {
}
}()
for {
+ // The next pass is due interval+jitter after this one STARTED, not
+ // after it finished: a pass is bounded by geoMirrorDeadline (10 min),
+ // so a change to the pointer is seen within interval+jitter (11 min)
+ // even when a long download was in flight (codex R8-2).
+ next := time.Now().Add(geoMirrorInterval + rand.N(geoMirrorJitter))
ctx, cancel := context.WithTimeout(context.Background(), geoMirrorDeadline)
m.Poll(ctx)
cancel()
select {
case <-stop:
return
- case <-time.After(geoMirrorInterval + rand.N(geoMirrorJitter)):
+ case <-time.After(time.Until(next)):
}
}
}
diff --git a/internal/edge/geoipmirror_test.go b/internal/edge/geoipmirror_test.go
index 6b473f0ae..7d0dc298c 100644
--- a/internal/edge/geoipmirror_test.go
+++ b/internal/edge/geoipmirror_test.go
@@ -289,3 +289,40 @@ func TestMirrorSyncFailureAfterRenameKeepsTheGeneration(t *testing.T) {
t.Fatal("restart after the failed sync did not recover the generation")
}
}
+
+// Codex R8-2: an off that lands during a download takes effect at the end
+// of that pass; the superseded download is never installed.
+func TestMirrorOffDuringADownloadNeverInstalls(t *testing.T) {
+ m, s, dir := newTestMirror(t)
+ publishFixture(t, s, geoFixture(t, "GeoIP2-City-Test.mmdb"), "GeoIP2-City", time.Now())
+ downloadHook = func() { s.Delete(context.Background(), geoip.PointerKey) } // off, mid-download
+ t.Cleanup(func() { downloadHook = nil })
+ if err := m.Poll(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ if m.sw.Lookup(net.ParseIP("81.2.69.142")).Loaded {
+ t.Fatal("a download superseded by off was installed")
+ }
+ if matches, _ := filepath.Glob(filepath.Join(dir, "*.mmdb")); len(matches) != 0 {
+ t.Fatalf("generations left: %v", matches)
+ }
+}
+
+// ...and a newer publish during a download installs the newer one in the
+// same pass.
+func TestMirrorNewerPublishDuringADownloadInstallsTheNewer(t *testing.T) {
+ m, s, _ := newTestMirror(t)
+ publishFixture(t, s, geoFixture(t, "GeoIP2-City-Test.mmdb"), "GeoIP2-City", time.Now())
+ var newer geoip.Pointer
+ downloadHook = func() {
+ downloadHook = nil // only the first download is interrupted
+ newer = publishFixture(t, s, geoFixture(t, "GeoIP2-Country-Test.mmdb"), "GeoIP2-Country", time.Now())
+ }
+ t.Cleanup(func() { downloadHook = nil })
+ if err := m.Poll(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ if st := m.State(); st.SHA256 != newer.SHA256 {
+ t.Fatalf("loaded %s, want the newer %s", st.SHA256, newer.SHA256)
+ }
+}