amadan

carlos / platform Public

console: geoip publish path (stage/commit, country probe, shrink check, blob+pointer, sweep, off, D1 expiry) (geoip spec §3, §4.1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 47ef0b17308bf880672d36911b753d2e37f2f0d4 parent 7d04b53
2 files changed, +516 −0
  • internal/console/geoippublish.go +291 −0
  • internal/console/geoippublish_test.go +225 −0
diff --git a/internal/console/geoippublish.go b/internal/console/geoippublish.go
new file mode 100644
index 000000000..f75668643
--- /dev/null
+++ b/internal/console/geoippublish.go
@@ -0,0 +1,291 @@
+package console
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "net"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/geo"
+ "github.com/carlosframework/platform/internal/geoip"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// geoipProbeAddrs is the country probe (spec §4.1 step 4): at least three
+// must answer with a country. A variable so tests can use addresses the
+// MaxMind fixtures hold (FACTS.md).
+var geoipProbeAddrs = []string{"8.8.8.8", "1.1.1.1", "9.9.9.9", "208.67.222.222", "2001:4860:4860::8888"}
+
+const geoipProbeNeed = 3
+
+var (
+ errGeoipBusy = errors.New("a geoip publish is already in progress")
+ errGeoipRefused = errors.New("geoip database refused")
+)
+
+func refused(category string) error { return fmt.Errorf("%w: %s", errGeoipRefused, category) }
+
+// geoipPublisher is the console's single writer of control/geoip/ (spec §3).
+// mu is held through cleanup by every publish, off, sweep and expiry, and a
+// publish that finds it held answers errGeoipBusy rather than queueing.
+type geoipPublisher struct {
+ mu sync.Mutex
+ store release.Store
+ tmpRoot string
+ maxBytes int64
+ deadline time.Duration
+ now func() time.Time
+}
+
+type geoipPublishOpts struct {
+ Origin, Edition string
+ Force bool
+}
+
+func newGeoipPublisher(store release.Store, tmpRoot string) *geoipPublisher {
+ return &geoipPublisher{store: store, tmpRoot: tmpRoot, maxBytes: geoip.MaxBlobBytes,
+ deadline: 15 * time.Minute, now: time.Now}
+}
+
+func marshalPointer(p geoip.Pointer) ([]byte, error) { return json.MarshalIndent(p, "", " ") }
+
+// 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)
+ if errors.Is(err, os.ErrNotExist) {
+ return geoip.Pointer{}, "", false, nil
+ }
+ if err != nil {
+ return geoip.Pointer{}, "", false, err
+ }
+ ptr, err := geoip.ParsePointer(raw)
+ if err != nil {
+ return geoip.Pointer{}, etag, true, err
+ }
+ return ptr, etag, true, nil
+}
+
+// geoipStaged is an upload on disk awaiting Commit. Whoever holds one owns
+// the publish slot (p.mu) and the temporary directory: Commit or Abandon
+// releases both (spec §4.2, codex R5-1).
+type geoipStaged struct {
+ dir, path, sha string
+ size int64
+}
+
+// Stage reserves the one publish slot BEFORE reading any of src, then
+// streams src to a temporary file, hashing and capping as it copies. On
+// success the slot stays held for Commit; on failure it is released.
+func (p *geoipPublisher) Stage(ctx context.Context, src io.Reader) (*geoipStaged, error) {
+ if !p.mu.TryLock() {
+ return nil, errGeoipBusy
+ }
+ dir, err := os.MkdirTemp(p.tmpRoot, "geoip-publish-*")
+ if err != nil {
+ p.mu.Unlock()
+ return nil, err
+ }
+ st := &geoipStaged{dir: dir, path: filepath.Join(dir, "db.mmdb")}
+ st.sha, st.size, err = copyCapped(ctx, st.path, src, p.maxBytes)
+ if err != nil {
+ p.Abandon(st)
+ return nil, err
+ }
+ return st, nil
+}
+
+// Abandon releases a staged upload that will not be committed.
+func (p *geoipPublisher) Abandon(st *geoipStaged) {
+ os.RemoveAll(st.dir)
+ p.mu.Unlock()
+}
+
+// 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)
+ defer cancel()
+ st, err := p.Stage(ctx, src)
+ if err != nil {
+ return geoip.Pointer{}, err
+ }
+ return p.Commit(ctx, st, o)
+}
+
+// Commit is spec §4.1 steps 3-6 on a staged upload. It always removes the
+// staged directory and releases the slot.
+func (p *geoipPublisher) Commit(ctx context.Context, st *geoipStaged, o geoipPublishOpts) (geoip.Pointer, error) {
+ defer p.Abandon(st)
+ if _, ok := ctx.Deadline(); !ok {
+ var cancel context.CancelFunc
+ ctx, cancel = context.WithTimeout(ctx, p.deadline)
+ defer cancel()
+ }
+ db, err := geo.OpenMMDB(st.path)
+ if err != nil {
+ return geoip.Pointer{}, refused("not a valid MaxMind-format database")
+ }
+ defer db.Close()
+ if !hasCountryData(db) {
+ return geoip.Pointer{}, refused("the database has no country data")
+ }
+ cur, etag, present, curErr := p.Current(ctx)
+ switch {
+ case curErr != nil && !present:
+ return geoip.Pointer{}, curErr // the store itself failed
+ case curErr != nil:
+ // A pointer exists but does not parse: replace it (etag still guards).
+ log.Printf("console: geoip publish: the current pointer is unreadable and will be replaced")
+ case present && !o.Force && cur.DatabaseType == db.Source() && db.NodeCount() < cur.NodeCount/2:
+ return geoip.Pointer{}, refused("fewer than half the current database's nodes (use -force if intended)")
+ }
+ next := geoip.Pointer{V: 1, SHA256: st.sha, Key: geoip.BlobKey(st.sha), Size: st.size,
+ DatabaseType: db.Source(), BuildEpoch: db.BuildEpoch().Unix(), NodeCount: db.NodeCount(),
+ Published: p.now().UTC().Truncate(time.Second), Origin: o.Origin, Edition: o.Edition}
+ f, err := os.Open(st.path)
+ if err != nil {
+ return geoip.Pointer{}, err
+ }
+ defer f.Close()
+ // A CAS conflict on the blob means a blob with this exact name, so these
+ // exact bytes, already exists: carry on (spec §3 step 2).
+ if _, err := release.PutObjectFrom(ctx, p.store, next.Key, f, st.size, ""); err != nil && !errors.Is(err, release.ErrCASConflict) {
+ return geoip.Pointer{}, err
+ }
+ body, err := marshalPointer(next)
+ if err != nil {
+ return geoip.Pointer{}, err
+ }
+ if !present {
+ etag = ""
+ }
+ if _, err := p.store.Put(ctx, geoip.PointerKey, body, etag); err != nil {
+ return geoip.Pointer{}, fmt.Errorf("pointer write: %w", err)
+ }
+ if err := p.sweepLocked(ctx, next.Key); err != nil {
+ log.Printf("console: geoip sweep after publish: %v", err)
+ }
+ log.Printf("console: geoip published %s (%s, %s)", st.sha[:8], next.DatabaseType, next.Origin)
+ return next, nil
+}
+
+// copyCapped streams src into path, hashing, refusing more than max bytes
+// as it copies (codex R1-7).
+func copyCapped(ctx context.Context, path string, src io.Reader, max int64) (string, int64, error) {
+ f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
+ if err != nil {
+ return "", 0, err
+ }
+ defer f.Close()
+ h := sha256.New()
+ n, err := io.Copy(io.MultiWriter(f, h), io.LimitReader(geoipCtxReader{ctx, src}, max+1))
+ if err != nil {
+ return "", 0, err
+ }
+ if n > max {
+ return "", 0, refused(fmt.Sprintf("larger than %d bytes", max))
+ }
+ if n == 0 {
+ return "", 0, refused("empty upload")
+ }
+ if err := f.Sync(); err != nil {
+ return "", 0, err
+ }
+ return hex.EncodeToString(h.Sum(nil)), n, nil
+}
+
+type geoipCtxReader struct {
+ ctx context.Context
+ r io.Reader
+}
+
+func (c geoipCtxReader) Read(b []byte) (int, error) {
+ if err := c.ctx.Err(); err != nil {
+ return 0, err
+ }
+ return c.r.Read(b)
+}
+
+func hasCountryData(db *geo.MMDB) bool {
+ n := 0
+ for _, a := range geoipProbeAddrs {
+ if r, ok := db.Lookup(net.ParseIP(a)); ok && r.Country != "" {
+ n++
+ }
+ }
+ return n >= geoipProbeNeed
+}
+
+// Off deletes the pointer, then every blob (spec §3). Unconditional delete:
+// safe under the single-writer mutex (codex R2-6).
+func (p *geoipPublisher) Off(ctx context.Context) error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if err := p.store.Delete(ctx, geoip.PointerKey); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ return p.sweepLocked(ctx, "")
+}
+
+// Sweep deletes every blob the pointer does not name (orphans from a crash
+// between blob upload and pointer write). Run at startup and daily.
+func (p *geoipPublisher) Sweep(ctx context.Context) error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ cur, _, present, err := p.Current(ctx)
+ if err != nil {
+ return err
+ }
+ keep := ""
+ if present && !cur.Expired {
+ keep = cur.Key
+ }
+ return p.sweepLocked(ctx, keep)
+}
+
+func (p *geoipPublisher) sweepLocked(ctx context.Context, keep string) error {
+ keys, err := p.store.List(ctx, geoip.Prefix)
+ if err != nil {
+ return err
+ }
+ var firstErr error
+ for _, k := range keys {
+ if !strings.HasSuffix(k, ".mmdb") || k == keep {
+ continue
+ }
+ if err := p.store.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) && firstErr == nil {
+ firstErr = err
+ }
+ }
+ return firstErr
+}
+
+// ExpireIfDue applies D1 to the bucket: a GeoLite2 pointer past expiry is
+// rewritten with expired=true and its blob destroyed.
+func (p *geoipPublisher) ExpireIfDue(ctx context.Context, now time.Time) (bool, error) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ cur, etag, present, err := p.Current(ctx)
+ if err != nil || !present || cur.Expired || !geoip.Expired(cur.DatabaseType, cur.Build(), now) {
+ return false, err
+ }
+ cur.Expired = true
+ body, err := marshalPointer(cur)
+ if err != nil {
+ return false, err
+ }
+ if _, err := p.store.Put(ctx, geoip.PointerKey, body, etag); err != nil {
+ return false, err
+ }
+ log.Printf("console: geoip database %s expired (GeoLite2, 30 days after build) and was removed", cur.SHA256[:8])
+ return true, p.sweepLocked(ctx, "")
+}
diff --git a/internal/console/geoippublish_test.go b/internal/console/geoippublish_test.go
new file mode 100644
index 000000000..106190a70
--- /dev/null
+++ b/internal/console/geoippublish_test.go
@@ -0,0 +1,225 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "os"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/geoip"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The repo's fixtures hold none of the production probe addresses
+// (FACTS.md), so tests swap in addresses the fixtures do resolve.
+func withFixtureProbes(t *testing.T) {
+ prev := geoipProbeAddrs
+ geoipProbeAddrs = []string{"81.2.69.142", "216.160.83.56", "89.160.20.112", "2001:218::1", "67.43.156.1"}
+ t.Cleanup(func() { geoipProbeAddrs = prev })
+}
+
+func fixture(t *testing.T, name string) []byte {
+ b, err := os.ReadFile("../geo/testdata/" + name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+}
+
+func newTestPublisher(t *testing.T) (*geoipPublisher, release.Store) {
+ s := release.NewDirStore(t.TempDir())
+ return newGeoipPublisher(s, t.TempDir()), s
+}
+
+func blobKeys(t *testing.T, s release.Store) []string {
+ keys, err := s.List(context.Background(), geoip.Prefix)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out []string
+ for _, k := range keys {
+ if strings.HasSuffix(k, ".mmdb") {
+ out = append(out, k)
+ }
+ }
+ return out
+}
+
+func TestPublishWritesBlobThenPointer(t *testing.T) {
+ withFixtureProbes(t)
+ p, s := newTestPublisher(t)
+ ptr, err := p.Publish(context.Background(), bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")), geoipPublishOpts{Origin: "manual"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if ptr.DatabaseType != "GeoIP2-City" || ptr.NodeCount != 1547 || ptr.Origin != "manual" || ptr.Key != geoip.BlobKey(ptr.SHA256) {
+ t.Fatalf("pointer = %+v", ptr)
+ }
+ raw, _, err := s.Get(context.Background(), geoip.PointerKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got, err := geoip.ParsePointer(raw); err != nil || got.SHA256 != ptr.SHA256 {
+ t.Fatalf("stored pointer = (%+v, %v)", got, err)
+ }
+ if keys := blobKeys(t, s); len(keys) != 1 || keys[0] != ptr.Key {
+ t.Fatalf("blobs = %v", keys)
+ }
+}
+
+func TestPublishRefusals(t *testing.T) {
+ withFixtureProbes(t)
+ ctx := context.Background()
+ p, _ := newTestPublisher(t)
+ for name, body := range map[string][]byte{
+ "not a database": []byte("junk"),
+ "no country data": fixture(t, "MaxMind-DB-test-ipv4-24.mmdb"),
+ } {
+ if _, err := p.Publish(ctx, bytes.NewReader(body), geoipPublishOpts{Origin: "manual"}); !errors.Is(err, errGeoipRefused) {
+ t.Errorf("%s: err = %v, want errGeoipRefused", name, err)
+ }
+ }
+}
+
+func TestPublishShrinkCheckOnlyForTheSameType(t *testing.T) {
+ withFixtureProbes(t)
+ ctx := context.Background()
+ p, s := newTestPublisher(t)
+ if _, err := p.Publish(ctx, bytes.NewReader(fixture(t, "GeoIP2-Country-Test.mmdb")), geoipPublishOpts{Origin: "manual"}); err != nil {
+ t.Fatal(err)
+ }
+ // Inflate the recorded node_count so a same-type republish looks like a
+ // truncation (fewer than half).
+ raw, etag, _ := s.Get(ctx, geoip.PointerKey)
+ inflated := bytes.Replace(raw, []byte(`"node_count": 1704`), []byte(`"node_count": 99999`), 1)
+ if bytes.Equal(inflated, raw) {
+ t.Fatalf("pointer JSON shape changed; adjust the test: %s", raw)
+ }
+ if _, err := s.Put(ctx, geoip.PointerKey, inflated, etag); err != nil {
+ t.Fatal(err)
+ }
+ same := fixture(t, "GeoIP2-Country-Test.mmdb")
+ if _, err := p.Publish(ctx, bytes.NewReader(same), geoipPublishOpts{Origin: "manual"}); !errors.Is(err, errGeoipRefused) {
+ t.Fatalf("same-type shrink = %v, want refused", err)
+ }
+ if _, err := p.Publish(ctx, bytes.NewReader(same), geoipPublishOpts{Origin: "manual", Force: true}); err != nil {
+ t.Fatalf("forced = %v", err)
+ }
+ // A different type skips the check entirely.
+ if _, err := p.Publish(ctx, bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")), geoipPublishOpts{Origin: "manual"}); err != nil {
+ t.Fatalf("different type = %v", err)
+ }
+}
+
+func TestPublishSameFileTwiceAndXYXKeepsTheCurrentBlob(t *testing.T) {
+ withFixtureProbes(t)
+ ctx := context.Background()
+ p, s := newTestPublisher(t)
+ x, y := fixture(t, "GeoIP2-City-Test.mmdb"), fixture(t, "GeoIP2-Country-Test.mmdb")
+ for i, body := range [][]byte{x, x, y, x} {
+ ptr, err := p.Publish(ctx, bytes.NewReader(body), geoipPublishOpts{Origin: "manual", Force: true})
+ if err != nil {
+ t.Fatalf("publish %d: %v", i, err)
+ }
+ keys := blobKeys(t, s)
+ if len(keys) != 1 || keys[0] != ptr.Key {
+ t.Fatalf("after publish %d blobs = %v, want only %s", i, keys, ptr.Key)
+ }
+ }
+}
+
+func TestPublishIsOneAtATime(t *testing.T) {
+ withFixtureProbes(t)
+ p, _ := newTestPublisher(t)
+ p.mu.Lock() // a publish in progress
+ _, err := p.Publish(context.Background(), bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")), geoipPublishOpts{Origin: "manual"})
+ p.mu.Unlock()
+ if !errors.Is(err, errGeoipBusy) {
+ t.Fatalf("concurrent publish = %v, want errGeoipBusy", err)
+ }
+}
+
+func TestPublishRefusesOversizeWhileCopying(t *testing.T) {
+ withFixtureProbes(t)
+ p, _ := newTestPublisher(t)
+ p.maxBytes = 1024
+ _, err := p.Publish(context.Background(), bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")), geoipPublishOpts{Origin: "manual"})
+ if !errors.Is(err, errGeoipRefused) {
+ t.Fatalf("oversize = %v", err)
+ }
+ if entries, _ := os.ReadDir(p.tmpRoot); len(entries) != 0 {
+ t.Fatalf("temp left behind: %v", entries)
+ }
+}
+
+func TestOffAndSweep(t *testing.T) {
+ withFixtureProbes(t)
+ ctx := context.Background()
+ p, s := newTestPublisher(t)
+ ptr, err := p.Publish(ctx, bytes.NewReader(fixture(t, "GeoIP2-City-Test.mmdb")), geoipPublishOpts{Origin: "manual"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ // An orphan from a crashed publish (spec §3 orphan sweep).
+ orphan := geoip.BlobKey(strings.Repeat("a", 64))
+ if _, err := s.Put(ctx, orphan, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.Sweep(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if keys := blobKeys(t, s); len(keys) != 1 || keys[0] != ptr.Key {
+ t.Fatalf("after sweep = %v", keys)
+ }
+ if err := p.Off(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := s.Get(ctx, geoip.PointerKey); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("pointer after off: %v", err)
+ }
+ if keys := blobKeys(t, s); len(keys) != 0 {
+ t.Fatalf("blobs after off = %v", keys)
+ }
+}
+
+func TestExpireIfDueMarksGeoLiteExpired(t *testing.T) {
+ ctx := context.Background()
+ p, s := newTestPublisher(t)
+ sha := strings.Repeat("b", 64)
+ ptr := geoip.Pointer{V: 1, SHA256: sha, Key: geoip.BlobKey(sha), Size: 1, DatabaseType: "GeoLite2-City",
+ BuildEpoch: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC).Unix(), Origin: "manual"}
+ putPointer(t, s, ptr)
+ if _, err := s.Put(ctx, ptr.Key, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if did, err := p.ExpireIfDue(ctx, time.Date(2026, 9, 20, 0, 0, 0, 0, time.UTC)); did || err != nil {
+ t.Fatalf("early = (%v, %v)", did, err)
+ }
+ if did, err := p.ExpireIfDue(ctx, time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)); !did || err != nil {
+ t.Fatalf("due = (%v, %v)", did, err)
+ }
+ got, _, present, err := p.Current(ctx)
+ if err != nil || !present || !got.Expired {
+ t.Fatalf("pointer = (%+v, %v, %v)", got, present, err)
+ }
+ if keys := blobKeys(t, s); len(keys) != 0 {
+ t.Fatalf("blob not destroyed: %v", keys)
+ }
+}
+
+func putPointer(t *testing.T, s release.Store, p geoip.Pointer) {
+ t.Helper()
+ b, err := marshalPointer(p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(context.Background(), geoip.PointerKey, b, ""); err != nil {
+ t.Fatal(err)
+ }
+}
+
+var _ = sync.Mutex{}