amadan

carlos / platform Public

geoip: keep every unsynced generation; feed status capped and written under its own deadline; feed-state blips keep the alert; pointer state tracked even with an unusable directory; geoip reason escalates while firing (Astra code review R2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude a10120deec108e9993c35febfac32720f3d111b1 parent 63da624
7 files changed, +198 −37
  • internal/console/geoipfeed.go +29 −20
  • internal/console/geoipfeed_test.go +29 −0
  • internal/console/geoippublish.go +8 −3
  • internal/edge/geoipmirror.go +39 −14
  • internal/edge/geoipmirror_test.go +61 −0
  • internal/metrics/alert.go +11 −0
  • internal/metrics/alert_test.go +21 −0

CI failed — run details

diff --git a/internal/console/geoipfeed.go b/internal/console/geoipfeed.go
index 1abeb6bf6..6f735cc34 100644
--- a/internal/console/geoipfeed.go
+++ b/internal/console/geoipfeed.go
@@ -97,7 +97,13 @@ var errFeedNotFound = errors.New("geoip feed: no edition available yet")
// Check is one feed check. It always writes feed.json.
func (f *geoipFeeder) Check(ctx context.Context) error {
- st := f.readStatus(ctx)
+ // Status I/O gets its own short deadline, detached from the check's: a
+ // check that ran out its 15 minutes on a stalled provider must still
+ // record the failure, or seven timeouts never add up to an alert (code
+ // review R2-2).
+ sctx, scancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+ defer scancel()
+ st, etag := f.readStatus(sctx)
st.V, st.Feed, st.Box, st.LastCheck = 1, f.spec.Name, f.box, f.now().UTC().Truncate(time.Second)
edition, err := f.check(ctx)
var ff geoipFeedFailure
@@ -122,7 +128,9 @@ func (f *geoipFeeder) Check(ctx context.Context) error {
st.LastFailure = "store"
log.Printf("console: geoip feed %s: check failed: store", f.spec.Name)
}
- f.writeStatus(ctx, st)
+ wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+ defer wcancel()
+ f.writeStatus(wctx, st, etag)
return err
}
@@ -133,33 +141,34 @@ func orNone(s string) string {
return s
}
-func (f *geoipFeeder) readStatus(ctx context.Context) geoip.FeedStatus {
- rc, _, err := release.OpenObject(ctx, f.store, geoip.FeedKey)
+// readStatus reads feed.json capped (never whole: code review R2-1) and
+// returns its etag for the conditional write. "" etag = absent.
+func (f *geoipFeeder) readStatus(ctx context.Context) (geoip.FeedStatus, string) {
+ b, etag, err := readCappedKey(ctx, f.store, geoip.FeedKey, geoip.MaxPointerBytes)
if err != nil {
- return geoip.FeedStatus{}
- }
- defer rc.Close()
- b, err := io.ReadAll(io.LimitReader(rc, geoip.MaxPointerBytes+1))
- if err != nil || len(b) > geoip.MaxPointerBytes {
- return geoip.FeedStatus{}
+ if errors.Is(err, errPointerTooLarge) {
+ return geoip.FeedStatus{}, etag // replace an oversized object, never decode it
+ }
+ return geoip.FeedStatus{}, ""
}
s, err := geoip.ParseFeedStatus(b)
if err != nil || s.Feed != f.spec.Name {
- return geoip.FeedStatus{} // another feed's history is not this one's
+ return geoip.FeedStatus{}, etag // another feed's history is not this one's
}
- return s
+ return s, etag
}
-func (f *geoipFeeder) writeStatus(ctx context.Context, st geoip.FeedStatus) {
+func (f *geoipFeeder) writeStatus(ctx context.Context, st geoip.FeedStatus, etag string) {
b, _ := json.MarshalIndent(st, "", " ")
- _, etag, err := f.store.Get(ctx, geoip.FeedKey)
- if errors.Is(err, os.ErrNotExist) {
- etag = ""
- } else if err != nil {
- log.Printf("console: geoip feed status unreadable; not written")
- return
- }
if _, err := f.store.Put(ctx, geoip.FeedKey, b, etag); err != nil {
+ if errors.Is(err, release.ErrCASConflict) && etag == "" {
+ // Present but unreadable before (oversized under DirStore, or a
+ // race): a delete then create-only put, best effort.
+ f.store.Delete(ctx, geoip.FeedKey)
+ if _, err := f.store.Put(ctx, geoip.FeedKey, b, ""); err == nil {
+ return
+ }
+ }
log.Printf("console: geoip feed status not written")
}
}
diff --git a/internal/console/geoipfeed_test.go b/internal/console/geoipfeed_test.go
index 094de712d..41551d7f4 100644
--- a/internal/console/geoipfeed_test.go
+++ b/internal/console/geoipfeed_test.go
@@ -280,3 +280,32 @@ func TestFeedSkipsWhileSomeoneIsPublishing(t *testing.T) {
t.Fatalf("a busy slot counted as a failure: %+v", st)
}
}
+
+// Code review R2-2: a check that used up its deadline still records the
+// failure.
+func TestFeedRecordsAFailureAfterItsDeadline(t *testing.T) {
+ f, s := newTestFeeder(t, "dbip", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ select {
+ case <-r.Context().Done():
+ case <-time.After(5 * time.Second):
+ }
+ }))
+ ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+ defer cancel()
+ if err := f.Check(ctx); err == nil {
+ t.Fatal("a stalled provider succeeded")
+ }
+ if st := feedStatus(t, s); st.ConsecutiveFailures != 1 || st.LastFailure != "download" {
+ t.Fatalf("feed.json after a timeout = %+v", st)
+ }
+}
+
+// Code review R2-1: an oversized feed.json is replaced, never read whole.
+func TestFeedReplacesAnOversizedStatus(t *testing.T) {
+ f, s := newTestFeeder(t, "dbip", http.NotFoundHandler())
+ s.Put(context.Background(), geoip.FeedKey, bytes.Repeat([]byte("x"), geoip.MaxPointerBytes+10), "")
+ f.Check(context.Background())
+ if st := feedStatus(t, s); st.ConsecutiveNotFound != 1 {
+ t.Fatalf("feed.json = %+v", st)
+ }
+}
diff --git a/internal/console/geoippublish.go b/internal/console/geoippublish.go
index a5bdc44aa..9f49f8dba 100644
--- a/internal/console/geoippublish.go
+++ b/internal/console/geoippublish.go
@@ -156,16 +156,21 @@ var errPointerTooLarge = errors.New("the object at the geoip pointer key is larg
// content's sha256 (release.DirStore, etagOf), so it is computed here from
// the bytes, which are complete whenever they are under the cap.
func readPointerCapped(ctx context.Context, s release.Store) ([]byte, string, error) {
- rc, etag, err := release.OpenObject(ctx, s, geoip.PointerKey)
+ return readCappedKey(ctx, s, geoip.PointerKey, geoip.MaxPointerBytes)
+}
+
+// readCappedKey is readPointerCapped for any small control object.
+func readCappedKey(ctx context.Context, s release.Store, key string, max int) ([]byte, string, error) {
+ rc, etag, err := release.OpenObject(ctx, s, key)
if err != nil {
return nil, "", err
}
defer rc.Close()
- b, err := io.ReadAll(io.LimitReader(rc, geoip.MaxPointerBytes+1))
+ b, err := io.ReadAll(io.LimitReader(rc, int64(max)+1))
if err != nil {
return nil, "", err
}
- if len(b) > geoip.MaxPointerBytes {
+ if len(b) > max {
if etag == "" {
// Without the store's etag a conditional replace cannot be
// aimed at it; only off (unconditional delete) clears it.
diff --git a/internal/edge/geoipmirror.go b/internal/edge/geoipmirror.go
index 58c4a1812..a764abf26 100644
--- a/internal/edge/geoipmirror.go
+++ b/internal/edge/geoipmirror.go
@@ -54,7 +54,8 @@ type geoMirror struct {
loadedBuild time.Time
skipSHA map[string]string // sha -> why (rejected | expired); cleared whenever the observed pointer changes
pointerSHA string // the sha the last readable pointer named; "" for none
- syncPending bool // the last `current` rename is not known durable: keep the previous generation
+ syncPending bool // the last `current` rename is not known durable
+ unsynced map[string]bool // every generation a crash could bring back while syncPending; kept until a sync succeeds
// The alert's inputs (geoip spec §10.4). The pointer's metadata is
// kept after the reader is dropped, so an expired pointer still reads
@@ -175,11 +176,11 @@ func (m *geoMirror) writeCurrentLocked(c geoCurrent) error {
// m.syncPending tells install to keep the previous generation too until
// a later write is durable (code review R1-4).
if err := syncDirFn(m.dir); err != nil {
- log.Printf("edge: geoip directory sync failed after selecting a generation; keeping the previous one until it succeeds")
+ log.Printf("edge: geoip directory sync failed after selecting a generation; keeping the earlier ones until it succeeds")
m.syncPending = true
return nil
}
- m.syncPending = false
+ m.syncPending, m.unsynced = false, nil
return nil
}
@@ -242,9 +243,9 @@ func (m *geoMirror) Poll(ctx context.Context) error { return m.poll(ctx, true) }
// 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)
- }
+ // The control-plane state (feed, pointer) is read whatever the local
+ // directory's health, so the alert sees a published database this box
+ // cannot install (code review R2-4).
m.readFeedStatus(ctx)
raw, err := m.readPointer(ctx)
var p geoip.Pointer
@@ -311,6 +312,9 @@ func (m *geoMirror) poll(ctx context.Context, recheck bool) error {
}
m.genMu.Unlock()
+ if err := os.MkdirAll(m.dir, 0o755); err != nil {
+ return m.fail("local directory unusable", err)
+ }
tmp, err := m.download(ctx, p)
if err == nil && !m.pointerStillNames(ctx, p.SHA256) {
os.Remove(tmp)
@@ -430,8 +434,21 @@ func (m *geoMirror) install(tmp string, p geoip.Pointer) error {
prev := m.loadedSHA
m.sw.Swap(db)
m.loadedSHA, m.loadedType, m.loadedBuild = p.SHA256, db.Source(), db.BuildEpoch()
- if m.syncPending && prev != "" {
- m.cleanLockedKeeping(false, p.SHA256, prev)
+ if m.syncPending {
+ // Until a sync succeeds, any `current` since the last durable one
+ // could come back after a crash: keep every one of them (code
+ // review R2 on R1-4).
+ if m.unsynced == nil {
+ m.unsynced = map[string]bool{}
+ }
+ if prev != "" {
+ m.unsynced[prev] = true
+ }
+ keep := []string{p.SHA256}
+ for sha := range m.unsynced {
+ keep = append(keep, sha)
+ }
+ m.cleanLockedKeeping(false, keep...)
} else {
m.cleanLocked(p.SHA256, false)
}
@@ -480,14 +497,22 @@ func (m *geoMirror) failLocked(category string, err error) error {
func (m *geoMirror) readFeedStatus(ctx context.Context) {
ctx, cancel := context.WithTimeout(ctx, geoPointerReadTimeout)
defer cancel()
- failing, box := false, ""
- if raw, err := readCapped(ctx, m.store, geoip.FeedKey, geoip.MaxPointerBytes); err == nil {
- if st, err := geoip.ParseFeedStatus(raw); err == nil {
- failing, box = st.Failing(), st.Box
- }
+ raw, err := readCapped(ctx, m.store, geoip.FeedKey, geoip.MaxPointerBytes)
+ if errors.Is(err, os.ErrNotExist) {
+ m.genMu.Lock()
+ m.feedFailing, m.feedBox = false, "" // no feed: confirmed
+ m.genMu.Unlock()
+ return
+ }
+ if err != nil {
+ return // keep the last known state: a blip must not clear a firing alert (code review R2-3)
+ }
+ st, err := geoip.ParseFeedStatus(raw)
+ if err != nil {
+ return
}
m.genMu.Lock()
- m.feedFailing, m.feedBox = failing, box
+ m.feedFailing, m.feedBox = st.Failing(), st.Box
m.genMu.Unlock()
}
diff --git a/internal/edge/geoipmirror_test.go b/internal/edge/geoipmirror_test.go
index 4e09507d3..6cdc389fe 100644
--- a/internal/edge/geoipmirror_test.go
+++ b/internal/edge/geoipmirror_test.go
@@ -435,3 +435,64 @@ func TestMirrorAlertFeedReasonOnlyOnTheConsoleBox(t *testing.T) {
t.Fatalf("another box = %v, want none", c)
}
}
+
+// Code review R2 on R1-4: while directory syncs keep failing, every
+// generation a crash could bring back is kept.
+func TestMirrorKeepsEveryUnsyncedGeneration(t *testing.T) {
+ m, s, dir := newTestMirror(t)
+ a := publishFixture(t, s, geoFixture(t, "GeoIP2-City-Test.mmdb"), "GeoIP2-City", time.Now())
+ m.Poll(context.Background()) // A, durable
+ syncDirFn = func(string) error { return errors.New("injected") }
+ t.Cleanup(func() { syncDirFn = syncDir })
+ b := publishFixture(t, s, geoFixture(t, "GeoIP2-Country-Test.mmdb"), "GeoIP2-Country", time.Now())
+ m.Poll(context.Background())
+ c := publishFixture(t, s, geoFixture(t, "MaxMind-DB-test-ipv4-24.mmdb"), "Test", time.Now())
+ m.Poll(context.Background()) // C will be refused? the mirror does not probe; it installs whatever verifies
+ for _, sha := range []string{a.SHA256, b.SHA256} {
+ if _, err := os.Stat(filepath.Join(dir, sha+".mmdb")); err != nil {
+ t.Fatalf("generation %s deleted while syncs keep failing", sha[:8])
+ }
+ }
+ _ = c
+ syncDirFn = syncDir
+ publishFixture(t, s, geoFixture(t, "GeoIP2-City-Test.mmdb"), "GeoIP2-City", time.Now().Add(time.Second))
+}
+
+// Code review R2-3: a read blip keeps the last known feed state.
+func TestMirrorFeedStateSurvivesAReadBlip(t *testing.T) {
+ now := time.Date(2026, 2, 10, 0, 0, 0, 0, time.UTC)
+ m, s, _ := newTestMirror(t)
+ m.now = func() time.Time { return now }
+ raw, _ := json.Marshal(geoip.FeedStatus{V: 1, Feed: "f", Box: "console-box", ConsecutiveFailures: 7})
+ s.Put(context.Background(), geoip.FeedKey, raw, "")
+ m.Poll(context.Background())
+ _, etag, _ := s.Get(context.Background(), geoip.FeedKey)
+ s.Put(context.Background(), geoip.FeedKey, []byte("garbled"), etag)
+ m.Poll(context.Background())
+ if c := m.AlertCode(now, "console-box"); c != GeoAlertFeed {
+ t.Fatalf("a garbled read cleared the feed reason: %v", c)
+ }
+ s.Delete(context.Background(), geoip.FeedKey)
+ m.Poll(context.Background())
+ if c := m.AlertCode(now, "console-box"); c != GeoAlertNone {
+ t.Fatalf("a deleted feed.json must clear it: %v", c)
+ }
+}
+
+// Code review R2-4: an unusable local directory still lets the alert see
+// the published pointer.
+func TestMirrorUnusableDirectoryStillTracksThePointer(t *testing.T) {
+ now := time.Date(2026, 2, 10, 0, 0, 0, 0, time.UTC)
+ s := release.NewDirStore(t.TempDir())
+ blocker := filepath.Join(t.TempDir(), "file")
+ os.WriteFile(blocker, []byte("x"), 0o644)
+ m := newGeoMirror(s, filepath.Join(blocker, "geoip"), geo.NewSwappable()) // a path under a file: MkdirAll fails
+ m.now = func() time.Time { return now }
+ publishFixture(t, s, geoFixture(t, "GeoIP2-City-Test.mmdb"), "GeoIP2-City", now)
+ if err := m.Poll(context.Background()); err == nil {
+ t.Fatal("an unusable directory reported success")
+ }
+ if c := m.AlertCode(now.Add(61*time.Minute), "box"); c != GeoAlertNotLoaded {
+ t.Fatalf("code = %v, want not loaded", c)
+ }
+}
diff --git a/internal/metrics/alert.go b/internal/metrics/alert.go
index c4a087500..24a404138 100644
--- a/internal/metrics/alert.go
+++ b/internal/metrics/alert.go
@@ -262,6 +262,17 @@ func Evaluate(r Rules, st AlertState, obs Observation) (AlertState, []Event) {
// geoip: a reason code against 0.5, so any reason fires after the
// sustain and a reading of 0 clears (geoip spec §10.4).
evalFrac("geoip", obs.GeoIP, 0.5, r.GeoIPSustain)
+ // A reason that changes WHILE firing (expiring becomes expired) is a
+ // new firing event with the new code: the alert stays up, its sustain
+ // is not restarted, and the operator is told what it now means (code
+ // review R2-5).
+ if obs.GeoIP != nil {
+ if cur := out.Rules["geoip"]; cur.Firing && *obs.GeoIP > 0.5 && cur.Value != *obs.GeoIP {
+ cur.Value = *obs.GeoIP
+ out.Rules["geoip"] = cur
+ emit("geoip", true, *obs.GeoIP)
+ }
+ }
return out, events
}
diff --git a/internal/metrics/alert_test.go b/internal/metrics/alert_test.go
index 4046d9cf9..e4f55fa2a 100644
--- a/internal/metrics/alert_test.go
+++ b/internal/metrics/alert_test.go
@@ -226,3 +226,24 @@ func TestGeoIPRuleFiresAfterItsSustainAndClearsOnZero(t *testing.T) {
t.Fatalf("a zero reading must clear at once: %v", ev)
}
}
+
+// Code review R2-5: a reason that changes while firing re-fires with the
+// new code, without clearing or restarting the sustain.
+func TestGeoIPReasonChangeWhileFiring(t *testing.T) {
+ r := DefaultRules()
+ t0 := time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC)
+ two, three := 2.0, 3.0
+ st := AlertState{V: 1, Box: "b"}
+ st, _ = Evaluate(r, st, Observation{At: t0, GeoIP: &two})
+ st, ev := Evaluate(r, st, Observation{At: t0.Add(30 * time.Minute), GeoIP: &two})
+ if len(ev) != 1 || ev[0].Value != 2 {
+ t.Fatalf("fire = %v", ev)
+ }
+ st, ev = Evaluate(r, st, Observation{At: t0.Add(31 * time.Minute), GeoIP: &three})
+ if len(ev) != 1 || !ev[0].Firing || ev[0].Value != 3 || !st.Rules["geoip"].Firing || st.Rules["geoip"].Value != 3 {
+ t.Fatalf("escalation = %v %+v", ev, st.Rules["geoip"])
+ }
+ if _, ev = Evaluate(r, st, Observation{At: t0.Add(32 * time.Minute), GeoIP: &three}); len(ev) != 0 {
+ t.Fatalf("an unchanged reason re-fired: %v", ev)
+ }
+}