amadan

carlos / platform Public

edge mirror: downloads survive expiry ticks; post-rename sync failure keeps the generation; unknown pointer version is off; rejections reset when the pointer moves; GeoIP armed before ready and in dev mode. release: DirStore.PutFrom never reads the existing object into memory (Astra plan R1-3/4/5/7/8/9)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b7af1b956c8c6a393bb8bb17ada5ee312f2c581a parent b00c58b
5 files changed, +198 −38
  • internal/edge/edge.go +20 −17
  • internal/edge/geoipmirror.go +49 −17
  • internal/edge/geoipmirror_test.go +97 −0
  • internal/geoip/pointer.go +5 −1
  • internal/release/store.go +27 −3
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 698aea30f..840c7ac5c 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -3028,6 +3028,26 @@ func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready fun
// deployment, where nothing is ever backhauled at all.
backhaulMaxInflight.Store(int64(backhaulInflightFromEnv()))
+ // GeoIP (geoip spec §5.1): a tunnel client takes the hub's answer and
+ // mirrors nothing; an edge with the deployment store mirrors; anything
+ // else has no database and only strips forged headers.
+ switch {
+ case e.geoTunnelClient:
+ e.geoMode = "tunnel"
+ case e.staticStore != nil:
+ e.geoMode = "mirror"
+ e.geoMirror = newGeoMirror(e.staticStore, filepath.Join(dataDir, "carlos", "geoip"), e.geo)
+ e.geoMirror.Boot()
+ default:
+ e.geoMode = "none"
+ }
+ if e.geoMirror != nil {
+ // Booted and started here, before ready(e) publishes the edge and
+ // before the dev-mode return, so the heartbeat never reads a
+ // half-set geoMode and a dev edge mirrors too (plan review R1-5).
+ go e.geoMirror.run(make(chan struct{})) // process-lifetime, like the cert mirror
+ }
+
go e.Watch(make(chan struct{})) // process-lifetime; never stopped
// Reload routes on SIGHUP too (an operator's explicit "now").
@@ -3124,20 +3144,6 @@ func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready fun
go hm.run(make(chan struct{})) // process-lifetime; never stopped, like the cert mirror's
}
- // GeoIP (geoip spec §5.1): a tunnel client takes the hub's answer and
- // mirrors nothing; an edge with the deployment store mirrors; anything
- // else has no database and only strips forged headers.
- switch {
- case e.geoTunnelClient:
- e.geoMode = "tunnel"
- case e.staticStore != nil:
- e.geoMode = "mirror"
- e.geoMirror = newGeoMirror(e.staticStore, filepath.Join(dataDir, "carlos", "geoip"), e.geo)
- e.geoMirror.Boot()
- default:
- e.geoMode = "none"
- }
-
// The edge-local door (door.go, #175 P7), armed right after the home map
// it dials through: its own key mirror, one synchronous refresh, its
// loop, then the socket. Off, with the reason in the journal, when there
@@ -3223,9 +3229,6 @@ func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready fun
// Started last: every field the loop reads (wake above) is in place.
go mirror.run(make(chan struct{})) // process-lifetime; never stopped
}
- if e.geoMirror != nil {
- go e.geoMirror.run(make(chan struct{})) // process-lifetime, like the cert mirror
- }
// The non-ACME :80 handler. See httpFallbackHandler's doc comment for the
// Allowed-vs-lookupUpstream split.
diff --git a/internal/edge/geoipmirror.go b/internal/edge/geoipmirror.go
index 015859ca6..5ff8a39ad 100644
--- a/internal/edge/geoipmirror.go
+++ b/internal/edge/geoipmirror.go
@@ -50,7 +50,8 @@ type geoMirror struct {
loadedSHA string
loadedType string
loadedBuild time.Time
- skipSHA map[string]string // sha -> why (rejected | expired); cleared when the pointer moves on
+ 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
lastErr string // last poll's error category, "" when healthy
failingAt time.Time
lastFailLog time.Time
@@ -108,7 +109,7 @@ func (m *geoMirror) Boot() {
keep = cur.SHA256
log.Printf("edge: geoip database %s (%s, built %s) loaded", short(cur.SHA256), db.Source(), db.BuildEpoch().Format("2006-01-02"))
}
- m.cleanLocked(keep)
+ m.cleanLocked(keep, true)
}
func short(sha string) string {
@@ -159,9 +160,18 @@ func (m *geoMirror) writeCurrentLocked(c geoCurrent) error {
if err := os.Rename(tmp.Name(), filepath.Join(m.dir, "current")); err != nil {
return err
}
- return syncDir(m.dir)
+ // 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.
+ if err := syncDirFn(m.dir); err != nil {
+ log.Printf("edge: geoip directory sync failed after selecting a generation; continuing")
+ }
+ return nil
}
+// syncDirFn is syncDir, a variable so a test can inject a failure.
+var syncDirFn = syncDir
+
func syncDir(dir string) error {
d, err := os.Open(dir)
if err != nil {
@@ -171,16 +181,18 @@ func syncDir(dir string) error {
return d.Sync()
}
-// cleanLocked deletes every generation but keep, every .incoming-* and
-// every stray .current-*.
-func (m *geoMirror) cleanLocked(keep string) {
+// cleanLocked deletes every generation but keep and every stray
+// .current-*. Downloads (.incoming-*) are deleted only at boot: at any other
+// 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) {
entries, _ := os.ReadDir(m.dir)
for _, e := range entries {
n := e.Name()
if n == "current" || (keep != "" && n == keep+".mmdb") {
continue
}
- if strings.HasSuffix(n, ".mmdb") || strings.HasPrefix(n, ".incoming-") || strings.HasPrefix(n, ".current-") {
+ if strings.HasSuffix(n, ".mmdb") || strings.HasPrefix(n, ".current-") || (atBoot && strings.HasPrefix(n, ".incoming-")) {
os.Remove(filepath.Join(m.dir, n))
}
}
@@ -194,7 +206,7 @@ func (m *geoMirror) clearLocked() {
}
m.sw.Swap(nil)
m.loadedSHA, m.loadedType, m.loadedBuild = "", "", time.Time{}
- m.cleanLocked("")
+ m.cleanLocked("", false)
}
// Poll is one pass (spec §5.3). The caller bounds ctx by geoMirrorDeadline.
@@ -203,29 +215,44 @@ func (m *geoMirror) Poll(ctx context.Context) error {
return m.fail("local directory unusable", err)
}
raw, err := readCapped(ctx, m.store, geoip.PointerKey, geoip.MaxPointerBytes)
- if errors.Is(err, os.ErrNotExist) {
+ var p geoip.Pointer
+ if err == nil {
+ p, err = geoip.ParsePointer(raw)
+ }
+ if errors.Is(err, os.ErrNotExist) || errors.Is(err, geoip.ErrUnknownVersion) {
+ // No pointer, or one this binary cannot read: GeoIP is off on this
+ // box (spec §3), never "keep what is loaded" (plan review R1-7).
m.genMu.Lock()
if m.loadedSHA != "" {
- log.Printf("edge: geoip turned off; no database published")
+ if errors.Is(err, geoip.ErrUnknownVersion) {
+ log.Printf("edge: geoip turned off; the published pointer has a version this binary cannot read")
+ } else {
+ log.Printf("edge: geoip turned off; no database published")
+ }
}
+ m.pointerSHA, m.skipSHA = "", map[string]string{}
m.clearLocked()
m.lastErr = ""
m.genMu.Unlock()
return nil
}
- if err != nil {
+ if raw == nil && err != nil {
return m.fail("pointer unreadable", err)
}
- p, err := geoip.ParsePointer(raw)
if err != nil {
return m.fail("pointer invalid", err)
}
m.genMu.Lock()
+ if p.SHA256 != m.pointerSHA {
+ // The pointer moved (to anything, including back to an sha seen
+ // before): every earlier rejection is void (plan review R1-8).
+ m.pointerSHA, m.skipSHA = p.SHA256, map[string]string{}
+ }
if p.Expired || geoip.Expired(p.DatabaseType, p.Build(), m.now()) {
if m.skipSHA[p.SHA256] != "expired" {
log.Printf("edge: geoip database %s expired (GeoLite2, 30 days after build) and was removed", short(p.SHA256))
}
- m.skipSHA = map[string]string{p.SHA256: "expired"}
+ m.skipSHA[p.SHA256] = "expired"
m.clearLocked()
m.lastErr = ""
m.genMu.Unlock()
@@ -236,12 +263,10 @@ func (m *geoMirror) Poll(ctx context.Context) error {
m.genMu.Unlock()
return nil
}
- if why, ok := m.skipSHA[p.SHA256]; ok {
+ if _, skip := m.skipSHA[p.SHA256]; skip {
m.genMu.Unlock()
- _ = why
return nil
}
- m.skipSHA = map[string]string{} // the pointer moved on
m.genMu.Unlock()
tmp, err := m.download(ctx, p)
@@ -257,6 +282,10 @@ func (m *geoMirror) Poll(ctx context.Context) error {
return m.install(tmp, p)
}
+// downloadHook is a test seam: called after .incoming is created, before
+// the body is copied, so a test can run an expiry tick mid-download.
+var downloadHook func()
+
var (
errGeoDigest = errors.New("digest mismatch")
errGeoSize = errors.New("size mismatch")
@@ -279,6 +308,9 @@ func (m *geoMirror) download(ctx context.Context, p geoip.Pointer) (string, erro
if m.maxBytes < limit {
limit = m.maxBytes
}
+ if downloadHook != nil {
+ downloadHook()
+ }
h := sha256.New()
n, err := io.Copy(io.MultiWriter(f, h), io.LimitReader(rc, limit+1))
if err == nil {
@@ -335,7 +367,7 @@ func (m *geoMirror) install(tmp string, p geoip.Pointer) error {
}
m.sw.Swap(db)
m.loadedSHA, m.loadedType, m.loadedBuild = p.SHA256, db.Source(), db.BuildEpoch()
- m.cleanLocked(p.SHA256)
+ 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 d4d39277c..6b473f0ae 100644
--- a/internal/edge/geoipmirror_test.go
+++ b/internal/edge/geoipmirror_test.go
@@ -5,6 +5,7 @@ import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
+ "errors"
"net"
"os"
"path/filepath"
@@ -192,3 +193,99 @@ func TestMirrorRefusesAnOversizeBlob(t *testing.T) {
t.Fatalf("incoming left: %v", matches)
}
}
+
+// Plan review R1-7: a pointer this binary cannot read turns GeoIP off.
+func TestMirrorUnknownPointerVersionTurnsOff(t *testing.T) {
+ m, s, _ := newTestMirror(t)
+ publishFixture(t, s, geoFixture(t, "GeoIP2-City-Test.mmdb"), "GeoIP2-City", time.Now())
+ if err := m.Poll(context.Background()); err != nil || !lookupGB(m) {
+ t.Fatalf("setup: %v", err)
+ }
+ ctx := context.Background()
+ raw, etag, _ := s.Get(ctx, geoip.PointerKey)
+ var p map[string]any
+ json.Unmarshal(raw, &p)
+ p["v"] = 2
+ raw2, _ := json.Marshal(p)
+ if _, err := s.Put(ctx, geoip.PointerKey, raw2, etag); err != nil {
+ t.Fatal(err)
+ }
+ if err := m.Poll(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if m.sw.Lookup(net.ParseIP("81.2.69.142")).Loaded {
+ t.Fatal("still serving under a pointer version this binary cannot read")
+ }
+}
+
+// Plan review R1-8: a rejection lasts only while the pointer names that sha.
+func TestMirrorRetriesARejectedShaAfterThePointerMoves(t *testing.T) {
+ m, s, _ := newTestMirror(t)
+ ctx := context.Background()
+ city := geoFixture(t, "GeoIP2-City-Test.mmdb")
+ b := publishFixture(t, s, city, "GeoIP2-City", time.Now())
+ good := append([]byte(nil), city...)
+ os.WriteFile(filepath.Join(testMirrorStoreDir[s], b.Key), append(city[:len(city)-1], 'x'), 0o644) // corrupt B
+ if err := m.Poll(ctx); err == nil {
+ t.Fatal("corrupt B installed")
+ }
+ publishFixture(t, s, geoFixture(t, "GeoIP2-Country-Test.mmdb"), "GeoIP2-Country", time.Now()) // A
+ if err := m.Poll(ctx); err != nil {
+ t.Fatal(err)
+ }
+ os.WriteFile(filepath.Join(testMirrorStoreDir[s], b.Key), good, 0o644) // B repaired
+ raw, _ := json.Marshal(b)
+ _, etag, _ := s.Get(ctx, geoip.PointerKey)
+ if _, err := s.Put(ctx, geoip.PointerKey, raw, etag); err != nil {
+ t.Fatal(err)
+ }
+ if err := m.Poll(ctx); err != nil {
+ t.Fatalf("repaired B after the pointer moved away and back: %v", err)
+ }
+ if st := m.State(); st.SHA256 != b.SHA256 {
+ t.Fatalf("loaded %s, want repaired B %s", st.SHA256, b.SHA256)
+ }
+}
+
+// Plan review R1-3: an expiry tick during a replacement download must not
+// unlink the download.
+func TestMirrorExpiryDuringADownloadKeepsTheDownload(t *testing.T) {
+ m, s, _ := newTestMirror(t)
+ now := time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC)
+ m.now = func() time.Time { return now }
+ publishFixture(t, s, geoFixture(t, "GeoIP2-City-Test.mmdb"), "GeoIP2-City", now)
+ m.Poll(context.Background())
+ m.genMu.Lock()
+ m.loadedType, m.loadedBuild = "GeoLite2-City", now.Add(-31*24*time.Hour) // A is now expired
+ m.genMu.Unlock()
+ publishFixture(t, s, geoFixture(t, "GeoIP2-Country-Test.mmdb"), "GeoIP2-Country", now) // B
+ downloadHook = func() { m.ExpireTick() }
+ t.Cleanup(func() { downloadHook = nil })
+ if err := m.Poll(context.Background()); err != nil {
+ t.Fatalf("replacement broken by a concurrent expiry: %v", err)
+ }
+ if !lookupGB(m) {
+ t.Fatal("B not serving")
+ }
+}
+
+// Plan review R1-4: a directory fsync failing after `current` is renamed
+// must not delete the generation `current` names.
+func TestMirrorSyncFailureAfterRenameKeepsTheGeneration(t *testing.T) {
+ m, s, dir := newTestMirror(t)
+ syncDirFn = func(string) error { return errors.New("injected") }
+ t.Cleanup(func() { syncDirFn = syncDir })
+ p := publishFixture(t, s, geoFixture(t, "GeoIP2-City-Test.mmdb"), "GeoIP2-City", time.Now())
+ if err := m.Poll(context.Background()); err != nil {
+ t.Fatalf("poll: %v", err)
+ }
+ if _, err := os.Stat(filepath.Join(dir, p.SHA256+".mmdb")); err != nil {
+ t.Fatal("the generation current names was deleted")
+ }
+ syncDirFn = syncDir
+ m2 := newGeoMirror(release.NewDirStore(t.TempDir()), dir, geo.NewSwappable())
+ m2.Boot()
+ if !lookupGB(m2) {
+ t.Fatal("restart after the failed sync did not recover the generation")
+ }
+}
diff --git a/internal/geoip/pointer.go b/internal/geoip/pointer.go
index 3cb5cb032..055915635 100644
--- a/internal/geoip/pointer.go
+++ b/internal/geoip/pointer.go
@@ -29,6 +29,10 @@ const (
// to sha (lower-case hex).
func BlobKey(sha string) string { return Prefix + sha + ".mmdb" }
+// ErrUnknownVersion is a pointer this binary cannot read. Spec §3: treat it
+// as no pointer (GeoIP off), never as "keep what is loaded".
+var ErrUnknownVersion = errors.New("unknown geoip pointer version")
+
// Pointer is control/geoip/current.json (spec §3).
type Pointer struct {
V int `json:"v"`
@@ -56,7 +60,7 @@ func ParsePointer(raw []byte) (Pointer, error) {
return Pointer{}, fmt.Errorf("geoip pointer: %w", err)
}
if p.V != 1 {
- return Pointer{}, fmt.Errorf("geoip pointer: unknown version %d", p.V)
+ return Pointer{}, fmt.Errorf("geoip pointer: version %d: %w", p.V, ErrUnknownVersion)
}
if !IsSHA256Hex(p.SHA256) {
return Pointer{}, errors.New("geoip pointer: sha256 is not 64 lower-case hex characters")
diff --git a/internal/release/store.go b/internal/release/store.go
index edb13047f..5607a6a8c 100644
--- a/internal/release/store.go
+++ b/internal/release/store.go
@@ -347,6 +347,20 @@ func (d *DirStore) path(key string) (string, error) {
return filepath.Join(d.root, filepath.FromSlash(key)), nil
}
+// hashFile is etagOf over a file, streamed.
+func hashFile(p string) (string, error) {
+ f, err := os.Open(p)
+ if err != nil {
+ return "", err
+ }
+ defer f.Close()
+ h := sha256.New()
+ if _, err := io.Copy(h, f); err != nil {
+ return "", err
+ }
+ return hex.EncodeToString(h.Sum(nil)), nil
+}
+
func etagOf(body []byte) string {
sum := sha256.Sum256(body)
return hex.EncodeToString(sum[:])
@@ -449,7 +463,9 @@ func (d *DirStore) PutFrom(_ context.Context, key string, body io.ReadSeeker, si
}
d.mu.Lock()
defer d.mu.Unlock()
- cur, err := os.ReadFile(p)
+ // The CAS gate without reading the existing object into memory (it may
+ // be 1 GiB): Stat for create-only, a streamed hash for replace-only.
+ _, err = os.Stat(p)
exists := err == nil
if err != nil && !errors.Is(err, fs.ErrNotExist) {
return "", err
@@ -457,8 +473,16 @@ func (d *DirStore) PutFrom(_ context.Context, key string, body io.ReadSeeker, si
if ifMatch == "" && exists {
return "", fmt.Errorf("dirstore: %s exists: %w", key, ErrCASConflict)
}
- if ifMatch != "" && (!exists || etagOf(cur) != ifMatch) {
- return "", fmt.Errorf("dirstore: %s moved on: %w", key, ErrCASConflict)
+ if ifMatch != "" {
+ cur := ""
+ if exists {
+ if cur, err = hashFile(p); err != nil {
+ return "", err
+ }
+ }
+ if !exists || cur != ifMatch {
+ return "", fmt.Errorf("dirstore: %s moved on: %w", key, ErrCASConflict)
+ }
}
if err := os.Chmod(tmp.Name(), 0o644); err != nil {
return "", err