amadan

carlos / platform Public

geoip: install errors never delete retained generations; the console follower expires the loaded reader whatever current says; an unreadable feed history is never overwritten (Astra code review R3)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e62e4fca05ce4a717a46da9120d1df0b892a3fd9 parent a10120d
6 files changed, +125 −14
  • internal/console/geoipfeed.go +20 −9
  • internal/console/geoipfeed_test.go +28 −0
  • internal/console/geoipfollower.go +17 −2
  • internal/console/geoipfollower_test.go +26 −0
  • internal/edge/geoipmirror.go +14 −3
  • internal/edge/geoipmirror_test.go +20 −0
diff --git a/internal/console/geoipfeed.go b/internal/console/geoipfeed.go
index 6f735cc34..8ca4ea41f 100644
--- a/internal/console/geoipfeed.go
+++ b/internal/console/geoipfeed.go
@@ -103,7 +103,7 @@ func (f *geoipFeeder) Check(ctx context.Context) error {
// review R2-2).
sctx, scancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
defer scancel()
- st, etag := f.readStatus(sctx)
+ st, etag, known := 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
@@ -128,6 +128,13 @@ func (f *geoipFeeder) Check(ctx context.Context) error {
st.LastFailure = "store"
log.Printf("console: geoip feed %s: check failed: store", f.spec.Name)
}
+ if !known {
+ // The existing history could not be read: overwriting it would turn
+ // seven recorded failures into one and clear a firing alert with no
+ // success behind it (code review R3-2). Next check, then.
+ log.Printf("console: geoip feed status unreadable; this check's outcome was not recorded")
+ return err
+ }
wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
defer wcancel()
f.writeStatus(wctx, st, etag)
@@ -143,19 +150,23 @@ func orNone(s string) string {
// 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) {
+// known=false means the existing object could not be read at all (a
+// transient store error): the caller must not overwrite it.
+func (f *geoipFeeder) readStatus(ctx context.Context) (st geoip.FeedStatus, etag string, known bool) {
b, etag, err := readCappedKey(ctx, f.store, geoip.FeedKey, geoip.MaxPointerBytes)
- if err != nil {
- if errors.Is(err, errPointerTooLarge) {
- return geoip.FeedStatus{}, etag // replace an oversized object, never decode it
- }
- return geoip.FeedStatus{}, ""
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ return geoip.FeedStatus{}, "", true // confirmed absent
+ case errors.Is(err, errPointerTooLarge):
+ return geoip.FeedStatus{}, etag, true // replace an oversized object, never decode it
+ case err != nil:
+ return geoip.FeedStatus{}, "", false
}
s, err := geoip.ParseFeedStatus(b)
if err != nil || s.Feed != f.spec.Name {
- return geoip.FeedStatus{}, etag // another feed's history is not this one's
+ return geoip.FeedStatus{}, etag, true // garbage, or another feed's history: replace it
}
- return s, etag
+ return s, etag, true
}
func (f *geoipFeeder) writeStatus(ctx context.Context, st geoip.FeedStatus, etag string) {
diff --git a/internal/console/geoipfeed_test.go b/internal/console/geoipfeed_test.go
index 41551d7f4..a5fec3f31 100644
--- a/internal/console/geoipfeed_test.go
+++ b/internal/console/geoipfeed_test.go
@@ -8,7 +8,9 @@ import (
"crypto/sha256"
"crypto/tls"
"encoding/hex"
+ "encoding/json"
"errors"
+ "io"
"net/http"
"net/http/httptest"
"os"
@@ -309,3 +311,29 @@ func TestFeedReplacesAnOversizedStatus(t *testing.T) {
t.Fatalf("feed.json = %+v", st)
}
}
+
+type flakyFeedStore struct {
+ release.Store
+ failOpen atomic.Bool
+}
+
+func (s *flakyFeedStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ if key == geoip.FeedKey && s.failOpen.Load() {
+ return nil, "", errors.New("transient")
+ }
+ return release.OpenObject(ctx, s.Store, key)
+}
+
+// Code review R3-2: an unreadable feed.json is never overwritten.
+func TestFeedNeverOverwritesAnUnreadableHistory(t *testing.T) {
+ f, s := newTestFeeder(t, "dbip", http.NotFoundHandler())
+ raw, _ := json.Marshal(geoip.FeedStatus{V: 1, Feed: f.spec.Name, Box: "b", ConsecutiveFailures: 7})
+ s.Put(context.Background(), geoip.FeedKey, raw, "")
+ flaky := &flakyFeedStore{Store: s}
+ flaky.failOpen.Store(true)
+ f.store = flaky
+ f.Check(context.Background())
+ if st := feedStatus(t, s); st.ConsecutiveFailures != 7 {
+ t.Fatalf("history rewritten after an unreadable read: %+v", st)
+ }
+}
diff --git a/internal/console/geoipfollower.go b/internal/console/geoipfollower.go
index 0d03aed30..eb6f75762 100644
--- a/internal/console/geoipfollower.go
+++ b/internal/console/geoipfollower.go
@@ -17,6 +17,10 @@ import (
"github.com/carlosframework/platform/internal/geoip"
)
+// followerExpired is D1's rule; a var so a test can apply it to the
+// GeoIP2 fixtures, which never expire.
+var followerExpired = geoip.Expired
+
type geoipFollower struct {
dir string
sw *geo.Swappable
@@ -36,6 +40,12 @@ func NewGeoipFollower(dir string, sw *geo.Swappable) *geoipFollower {
func (f *geoipFollower) Tick() {
f.mu.Lock()
defer f.mu.Unlock()
+ // D1 on what is LOADED, first and whatever the local state says: an
+ // unreadable or unopenable `current` must never keep an expired reader
+ // serving (code review R3-1).
+ if src, build, ok := f.sw.Loaded(); ok && followerExpired(src, build, f.now()) {
+ f.drop("console: geoip database expired (GeoLite2, 30 days after build) and was closed")
+ }
raw, err := os.ReadFile(filepath.Join(f.dir, geoip.LocalCurrentFile))
var cur geoip.LocalCurrent
if err == nil {
@@ -48,7 +58,7 @@ func (f *geoipFollower) Tick() {
case err != nil:
f.logHourly("console: geoip local state unreadable; keeping the loaded database")
return
- case geoip.Expired(cur.DatabaseType, time.Unix(cur.BuildEpoch, 0), f.now()):
+ case followerExpired(cur.DatabaseType, time.Unix(cur.BuildEpoch, 0), f.now()):
f.drop("console: geoip database expired (GeoLite2, 30 days after build) and was closed")
return
case cur.SHA256 == f.loadedSHA:
@@ -60,13 +70,18 @@ func (f *geoipFollower) Tick() {
f.logHourly("console: geoip generation named by the edge is unopenable; keeping the loaded database")
return
}
+ if followerExpired(db.Source(), db.BuildEpoch(), f.now()) {
+ db.Close() // the file's own metadata, not just what `current` claims
+ f.drop("console: geoip database expired (GeoLite2, 30 days after build) and was closed")
+ return
+ }
f.sw.Swap(db)
f.loadedSHA = cur.SHA256
log.Printf("console: geoip database %s (%s) followed from this box's edge", cur.SHA256[:8], db.Source())
}
func (f *geoipFollower) drop(msg string) {
- if f.loadedSHA == "" {
+ if _, _, loaded := f.sw.Loaded(); f.loadedSHA == "" && !loaded {
return
}
f.sw.Swap(nil)
diff --git a/internal/console/geoipfollower_test.go b/internal/console/geoipfollower_test.go
index b223cf8b7..b30468ad1 100644
--- a/internal/console/geoipfollower_test.go
+++ b/internal/console/geoipfollower_test.go
@@ -90,3 +90,29 @@ func TestFollowerKeepsTheDatabaseOnAnUnreadableState(t *testing.T) {
t.Fatal("a garbled current dropped the database")
}
}
+
+// Code review R3-1: the loaded reader's own expiry is checked every tick,
+// even when `current` is unreadable or names an unopenable generation.
+func TestFollowerExpiresTheLoadedReaderWhateverCurrentSays(t *testing.T) {
+ for name, breakCurrent := range map[string]func(dir string){
+ "unreadable current": func(dir string) {
+ os.WriteFile(filepath.Join(dir, geoip.LocalCurrentFile), []byte("garbled"), 0o644)
+ },
+ "unopenable candidate": func(dir string) {
+ writeLocal(t, dir, strings.Repeat("e", 64), []byte("not a database"), "GeoIP2-City", time.Now())
+ },
+ } {
+ dir := filepath.Join(t.TempDir(), "geoip")
+ sw := geo.NewSwappable()
+ f := NewGeoipFollower(dir, sw)
+ writeLocal(t, dir, strings.Repeat("d", 64), fixture(t, "GeoIP2-City-Test.mmdb"), "GeoIP2-City", time.Now())
+ f.Tick()
+ breakCurrent(dir)
+ followerExpired = func(string, time.Time, time.Time) bool { return true }
+ f.Tick()
+ followerExpired = geoip.Expired
+ if sw.Lookup(net.ParseIP("81.2.69.142")).Loaded {
+ t.Errorf("%s: an expired reader kept serving", name)
+ }
+ }
+}
diff --git a/internal/edge/geoipmirror.go b/internal/edge/geoipmirror.go
index a764abf26..e7b32c839 100644
--- a/internal/edge/geoipmirror.go
+++ b/internal/edge/geoipmirror.go
@@ -414,21 +414,21 @@ func (m *geoMirror) install(tmp string, p geoip.Pointer) error {
syncDir(m.dir)
db, err := geo.OpenMMDBTrusted(final)
if err != nil {
- os.Remove(final)
+ m.removeGenerationLocked(p.SHA256)
m.skipSHA[p.SHA256] = "rejected"
log.Printf("edge: geoip blob %s rejected: unopenable", short(p.SHA256))
return m.failLocked("blob unopenable", err)
}
if geoip.Expired(db.Source(), db.BuildEpoch(), m.now()) {
db.Close()
- os.Remove(final)
+ m.removeGenerationLocked(p.SHA256)
m.skipSHA[p.SHA256] = "expired"
log.Printf("edge: geoip database %s expired (GeoLite2, 30 days after build) and was removed", short(p.SHA256))
return nil
}
if err := m.writeCurrentLocked(geoCurrent{SHA256: p.SHA256, DatabaseType: db.Source(), BuildEpoch: db.BuildEpoch().Unix()}); err != nil {
db.Close()
- os.Remove(final)
+ m.removeGenerationLocked(p.SHA256)
return m.failLocked("install failed", err)
}
prev := m.loadedSHA
@@ -457,6 +457,17 @@ func (m *geoMirror) install(tmp string, p geoip.Pointer) error {
return nil
}
+// removeGenerationLocked deletes a failed install's file unless the same
+// generation is one a crash could bring back (loaded, or kept while syncs
+// fail): republishing a retained A must never delete A on an error path
+// (code review R3 on R1-4).
+func (m *geoMirror) removeGenerationLocked(sha string) {
+ if sha == m.loadedSHA || m.unsynced[sha] {
+ return
+ }
+ os.Remove(m.path(sha))
+}
+
// ExpireTick is D1 on the loaded generation, under genMu, so it can only
// ever remove the generation that is current at that moment (codex R3-1).
func (m *geoMirror) ExpireTick() {
diff --git a/internal/edge/geoipmirror_test.go b/internal/edge/geoipmirror_test.go
index 6cdc389fe..4abb18244 100644
--- a/internal/edge/geoipmirror_test.go
+++ b/internal/edge/geoipmirror_test.go
@@ -496,3 +496,23 @@ func TestMirrorUnusableDirectoryStillTracksThePointer(t *testing.T) {
t.Fatalf("code = %v, want not loaded", c)
}
}
+
+// Code review R3 on R1-4: an install error path never deletes a generation
+// a crash could bring back.
+func TestRemoveGenerationKeepsRetainedOnes(t *testing.T) {
+ m, _, dir := newTestMirror(t)
+ os.MkdirAll(dir, 0o755)
+ a, b, c := strings.Repeat("a", 64), strings.Repeat("b", 64), strings.Repeat("c", 64)
+ for _, sha := range []string{a, b, c} {
+ os.WriteFile(filepath.Join(dir, sha+".mmdb"), []byte("x"), 0o644)
+ }
+ m.loadedSHA, m.unsynced = a, map[string]bool{b: true}
+ for _, sha := range []string{a, b, c} {
+ m.removeGenerationLocked(sha)
+ }
+ for sha, want := range map[string]bool{a: true, b: true, c: false} {
+ if _, err := os.Stat(filepath.Join(dir, sha+".mmdb")); (err == nil) != want {
+ t.Errorf("%s present=%v, want %v", sha[:1], err == nil, want)
+ }
+ }
+}