amadan

carlos / platform Public

geoip: capped console pointer reads; keep the previous generation until a selection is durable; Stage never reads a refused body; other row sorts last (Astra code review R1-3..6)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 9f501921170e75f79e334c83e556adc40a8f5764 parent f83276b
7 files changed, +136 −17
  • internal/console/analyticsview.go +14 −11
  • internal/console/geoippublish.go +38 −1
  • internal/console/geoippublish_test.go +43 −0
  • internal/console/server.go +1 −0
  • internal/console/templates/app_analytics.html +1 −1
  • internal/edge/geoipmirror.go +27 −4
  • internal/edge/geoipmirror_test.go +12 −0

CI passed — run details

diff --git a/internal/console/analyticsview.go b/internal/console/analyticsview.go
index f4fcb6158..2f895ec4e 100644
--- a/internal/console/analyticsview.go
+++ b/internal/console/analyticsview.go
@@ -416,17 +416,20 @@ func geoCountries(m map[string]int64) map[string]int64 {
if len(m) == 0 {
return nil
}
- out := map[string]int64{}
- for k, v := range topWithOther(m, 10) {
- switch k {
- case "_unknown":
- k = "unknown"
- case "_other":
- k = "other"
- }
- out[k] += v
- }
- return out
+ // The reserved keys stay as they are so sortedKV can put _other last;
+ // geoLabel gives them words at render time (code review R1-6).
+ return topWithOther(m, 10)
+}
+
+// geoLabel is a Countries row's label: the reserved buckets get words.
+func geoLabel(k string) string {
+ switch k {
+ case "_unknown":
+ return "unknown"
+ case "_other":
+ return "other"
+ }
+ return k
}
// geoAttributions is one credit line per source in the range that requires
diff --git a/internal/console/geoippublish.go b/internal/console/geoippublish.go
index e56c83e5c..070091bcd 100644
--- a/internal/console/geoippublish.go
+++ b/internal/console/geoippublish.go
@@ -75,10 +75,14 @@ func marshalPointer(p geoip.Pointer) ([]byte, error) { return json.MarshalIndent
// Current reads the pointer. present=false means no pointer (feature off).
func (p *geoipPublisher) Current(ctx context.Context) (geoip.Pointer, string, bool, error) {
- raw, etag, err := p.store.Get(ctx, geoip.PointerKey)
+ raw, etag, err := readPointerCapped(ctx, p.store)
if errors.Is(err, os.ErrNotExist) {
return geoip.Pointer{}, "", false, nil
}
+ if errors.Is(err, errPointerTooLarge) {
+ // Present but not a pointer anyone wrote: replaceable, never decoded.
+ return geoip.Pointer{}, etag, true, err
+ }
if err != nil {
return geoip.Pointer{}, "", false, err
}
@@ -137,6 +141,39 @@ func (p *geoipPublisher) Abandon(st *geoipStaged) {
p.mu.Unlock()
}
+var errPointerTooLarge = errors.New("the object at the geoip pointer key is larger than a pointer can be")
+
+// readPointerCapped reads control/geoip/current.json streamed and capped at
+// geoip.MaxPointerBytes, so an oversized object there can never be pulled
+// into the console's memory (code review R1-3). The etag is the store's;
+// DirStore cannot report one from Open, and its etag is defined as the
+// 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)
+ if err != nil {
+ return nil, "", err
+ }
+ defer rc.Close()
+ b, err := io.ReadAll(io.LimitReader(rc, geoip.MaxPointerBytes+1))
+ if err != nil {
+ return nil, "", err
+ }
+ if len(b) > geoip.MaxPointerBytes {
+ if etag == "" {
+ // Without the store's etag a conditional replace cannot be
+ // aimed at it; only off (unconditional delete) clears it.
+ return nil, "", errPointerTooLarge
+ }
+ return nil, etag, errPointerTooLarge
+ }
+ if etag == "" {
+ sum := sha256.Sum256(b)
+ etag = hex.EncodeToString(sum[:])
+ }
+ return b, etag, nil
+}
+
// Publish is Stage then Commit, under one deadline.
func (p *geoipPublisher) Publish(ctx context.Context, src io.Reader, o geoipPublishOpts) (geoip.Pointer, error) {
ctx, cancel := context.WithTimeout(ctx, p.deadline)
diff --git a/internal/console/geoippublish_test.go b/internal/console/geoippublish_test.go
index 6d93cf327..1563bc46f 100644
--- a/internal/console/geoippublish_test.go
+++ b/internal/console/geoippublish_test.go
@@ -6,6 +6,7 @@ import (
"crypto/sha256"
"encoding/hex"
"errors"
+ "fmt"
"os"
"path/filepath"
"strings"
@@ -350,3 +351,45 @@ func TestOffRefusesDuringAPublishThenSucceeds(t *testing.T) {
t.Fatalf("blobs after off = %v", keys)
}
}
+
+// Code review R1-3: an oversized object at the pointer key is never read
+// whole, and is reported as present-but-not-a-pointer.
+func TestCurrentRefusesAnOversizedPointer(t *testing.T) {
+ ctx := context.Background()
+ p, s := newTestPublisher(t)
+ if _, err := s.Put(ctx, geoip.PointerKey, bytes.Repeat([]byte("x"), geoip.MaxPointerBytes+1), ""); err != nil {
+ t.Fatal(err)
+ }
+ _, etag, present, err := p.Current(ctx)
+ if !errors.Is(err, errPointerTooLarge) || !present {
+ t.Fatalf("Current = (%v, %v), want errPointerTooLarge, present", present, err)
+ }
+ _ = etag
+}
+
+type panicReader struct{}
+
+func (panicReader) Read([]byte) (int, error) { panic("a busy Stage read its body") }
+
+// Code review R1-5: a Stage refused for a held slot never touches its body.
+func TestStageNeverReadsWhenTheSlotIsHeld(t *testing.T) {
+ p, _ := newTestPublisher(t)
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if _, err := p.Stage(context.Background(), panicReader{}); !errors.Is(err, errGeoipBusy) {
+ t.Fatalf("Stage = %v, want errGeoipBusy", err)
+ }
+}
+
+// Code review R1-6: the other row sorts last whatever its size.
+func TestGeoCountriesKeepsOtherLast(t *testing.T) {
+ m := map[string]int64{}
+ for i := 0; i < 15; i++ {
+ m[fmt.Sprintf("C%02d", i)] = 10
+ }
+ m["GB"] = 11
+ rows := sortedKV(geoCountries(m))
+ if rows[len(rows)-1].Key != "_other" || geoLabel(rows[len(rows)-1].Key) != "other" {
+ t.Fatalf("last row = %+v, want _other rendered as other", rows[len(rows)-1])
+ }
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index 9c2d267c0..cae913f1b 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -143,6 +143,7 @@ var templates = template.Must(template.New("").Funcs(template.FuncMap{
// geoCountries and geoAttributions: the Analytics tab's Countries table
// and the credit its sources require (geoip spec §9, §13).
"geoCountries": geoCountries,
+ "geoLabel": geoLabel,
"geoAttributions": geoAttributions,
// bytesOrDash renders a byte count in the smallest unit that keeps it
// readable; nil (no reading, from a unit that reported CPU but not
diff --git a/internal/console/templates/app_analytics.html b/internal/console/templates/app_analytics.html
index d1e9d3656..e44adf352 100644
--- a/internal/console/templates/app_analytics.html
+++ b/internal/console/templates/app_analytics.html
@@ -41,7 +41,7 @@ account on the <a href="{{.Base}}/apps/{{.AppName}}/settings">Settings</a> tab.<
<h3>Named events{{if .Sum.EventsApprox}} <span class="small">approximate</span>{{end}}</h3>
{{template "kvtable" .Sum.Events}}
<h3>Countries</h3>
-{{with geoCountries .Sum.Countries}}{{template "kvtable" .}}{{else}}<p class="small">No country data for this period.</p>{{end}}
+{{with geoCountries .Sum.Countries}}<div class="tablewrap"><table>{{range sortedKV .}}<tr><td>{{geoLabel .Key}}</td><td>{{.Value}}</td></tr>{{end}}</table></div>{{else}}<p class="small">No country data for this period.</p>{{end}}
{{range geoAttributions .Sum.GeoSources}}<p class="small"><a href="{{.URL}}">{{.Text}}</a></p>{{end}}
{{end}}
<p class="small">Times are UTC. The same view from a terminal: <code>carlos analytics --app {{.AppName}} --range {{.AnalyticsRange}}</code>.</p>
diff --git a/internal/edge/geoipmirror.go b/internal/edge/geoipmirror.go
index 419645b3e..535684185 100644
--- a/internal/edge/geoipmirror.go
+++ b/internal/edge/geoipmirror.go
@@ -56,6 +56,7 @@ 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
lastErr string // last poll's error category, "" when healthy
failingAt time.Time
lastFailLog time.Time
@@ -166,10 +167,16 @@ func (m *geoMirror) writeCurrentLocked(c geoCurrent) error {
}
// After the rename `current` already names the new generation, so a
// failed directory fsync must not make the caller delete it (plan review
- // R1-4): it is logged, and the rename stands.
+ // R1-4): it is logged, and the rename stands. But the rename is not
+ // durable either: a crash could bring back the PREVIOUS `current`, so
+ // 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; continuing")
+ log.Printf("edge: geoip directory sync failed after selecting a generation; keeping the previous one until it succeeds")
+ m.syncPending = true
+ return nil
}
+ m.syncPending = false
return nil
}
@@ -190,10 +197,21 @@ func syncDir(dir string) error {
// time one may be in progress outside genMu, and unlinking it would break
// a replacement that an expiry tick happened to race (plan review R1-3).
func (m *geoMirror) cleanLocked(keep string, atBoot bool) {
+ m.cleanLockedKeeping(atBoot, keep)
+}
+
+// cleanLockedKeeping is cleanLocked with any number of generations kept.
+func (m *geoMirror) cleanLockedKeeping(atBoot bool, keep ...string) {
+ kept := map[string]bool{}
+ for _, k := range keep {
+ if k != "" {
+ kept[k+".mmdb"] = true
+ }
+ }
entries, _ := os.ReadDir(m.dir)
for _, e := range entries {
n := e.Name()
- if n == "current" || (keep != "" && n == keep+".mmdb") {
+ if n == "current" || kept[n] {
continue
}
if strings.HasSuffix(n, ".mmdb") || strings.HasPrefix(n, ".current-") || (atBoot && strings.HasPrefix(n, ".incoming-")) {
@@ -395,9 +413,14 @@ func (m *geoMirror) install(tmp string, p geoip.Pointer) error {
os.Remove(final)
return m.failLocked("install failed", err)
}
+ prev := m.loadedSHA
m.sw.Swap(db)
m.loadedSHA, m.loadedType, m.loadedBuild = p.SHA256, db.Source(), db.BuildEpoch()
- m.cleanLocked(p.SHA256, false)
+ if m.syncPending && prev != "" {
+ m.cleanLockedKeeping(false, p.SHA256, prev)
+ } else {
+ m.cleanLocked(p.SHA256, false)
+ }
m.lastErr, m.failingAt = "", time.Time{}
log.Printf("edge: geoip database %s (%s, built %s) loaded", short(p.SHA256), db.Source(), db.BuildEpoch().Format("2006-01-02"))
return nil
diff --git a/internal/edge/geoipmirror_test.go b/internal/edge/geoipmirror_test.go
index 7d0dc298c..c8fed921d 100644
--- a/internal/edge/geoipmirror_test.go
+++ b/internal/edge/geoipmirror_test.go
@@ -282,6 +282,18 @@ func TestMirrorSyncFailureAfterRenameKeepsTheGeneration(t *testing.T) {
if _, err := os.Stat(filepath.Join(dir, p.SHA256+".mmdb")); err != nil {
t.Fatal("the generation current names was deleted")
}
+ // A second install while the first rename is still not durable keeps
+ // BOTH generations: a crash could bring back either `current` (code
+ // review R1-4).
+ q := publishFixture(t, s, geoFixture(t, "GeoIP2-Country-Test.mmdb"), "GeoIP2-Country", time.Now())
+ if err := m.Poll(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ for _, sha := range []string{p.SHA256, q.SHA256} {
+ if _, err := os.Stat(filepath.Join(dir, sha+".mmdb")); err != nil {
+ t.Fatalf("generation %s deleted while the selection was not durable", sha[:8])
+ }
+ }
syncDirFn = syncDir
m2 := newGeoMirror(release.NewDirStore(t.TempDir()), dir, geo.NewSwappable())
m2.Boot()