amadan

carlos / platform Public

edge: add local-only per-instance purge listener

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Paul Campbell 16d329c38c6c411b02a4f8e40f555ba5ab4912a2 parent 9ead6e1
2 files changed, +173 −0
  • internal/edge/purge.go +64 −0
  • internal/edge/purge_test.go +109 −0
diff --git a/internal/edge/purge.go b/internal/edge/purge.go
new file mode 100644
index 000000000..8ee360962
--- /dev/null
+++ b/internal/edge/purge.go
@@ -0,0 +1,64 @@
+package edge
+
+import (
+ "encoding/json"
+ "errors"
+ "net"
+ "net/http"
+ "os"
+)
+
+// PurgeListener is a per-instance, local-only channel for invalidating
+// that instance's cache entries right after a write — see
+// docs/superpowers/specs/2026-08-02-edge-cache-before-wake-design.md §3-4.
+// It is deliberately NOT reachable over the network: an externally
+// reachable purge endpoint would let anyone manufacture wake storms by
+// force-purging popular pages, exactly what this feature exists to
+// prevent.
+type PurgeListener struct {
+ ln net.Listener
+ srv *http.Server
+}
+
+type purgeRequest struct {
+ Path string `json:"path"`
+ Query string `json:"query"`
+}
+
+// ListenPurge binds socketPath (removing any stale file left behind by a
+// crashed previous instance) and serves purge requests scoped to host —
+// host is fixed here, never taken from the request, so a compromised or
+// buggy instance can only ever purge its own cache entries.
+func ListenPurge(socketPath, host string, cache *Cache) (*PurgeListener, error) {
+ if err := os.Remove(socketPath); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return nil, err
+ }
+ ln, err := net.Listen("unix", socketPath)
+ if err != nil {
+ return nil, err
+ }
+ if err := os.Chmod(socketPath, 0600); err != nil {
+ ln.Close()
+ return nil, err
+ }
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("POST /purge", func(w http.ResponseWriter, r *http.Request) {
+ var req purgeRequest
+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+ http.Error(w, "bad request", http.StatusBadRequest)
+ return
+ }
+ cache.PurgeURL(host, req.Path, req.Query)
+ w.WriteHeader(http.StatusOK)
+ })
+
+ srv := &http.Server{Handler: mux}
+ p := &PurgeListener{ln: ln, srv: srv}
+ go srv.Serve(ln) //nolint:errcheck // Close() below triggers the expected ErrServerClosed
+ return p, nil
+}
+
+func (p *PurgeListener) Close() error {
+ return p.srv.Close()
+}
diff --git a/internal/edge/purge_test.go b/internal/edge/purge_test.go
new file mode 100644
index 000000000..485413f26
--- /dev/null
+++ b/internal/edge/purge_test.go
@@ -0,0 +1,109 @@
+package edge
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "net"
+ "net/http"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+)
+
+func dialPurgeSocket(t *testing.T, socketPath string) *http.Client {
+ t.Helper()
+ return &http.Client{
+ Transport: &http.Transport{
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return net.Dial("unix", socketPath)
+ },
+ },
+ }
+}
+
+func TestListenPurgeRemovesCachedEntry(t *testing.T) {
+ dir := t.TempDir()
+ sock := filepath.Join(dir, "instance.sock")
+ cache := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
+ cache.Put(CacheKey{Host: "a.example", Method: "GET", Path: "/event"}, &CachedResponse{
+ StatusCode: 200, Header: http.Header{}, Body: []byte("hi"), StoredAt: time.Now(), TTL: time.Minute,
+ })
+
+ pl, err := ListenPurge(sock, "a.example", cache)
+ if err != nil {
+ t.Fatalf("ListenPurge: %v", err)
+ }
+ defer pl.Close()
+
+ body, _ := json.Marshal(map[string]string{"path": "/event", "query": ""})
+ client := dialPurgeSocket(t, sock)
+ resp, err := client.Post("http://unix/purge", "application/json", bytes.NewReader(body))
+ if err != nil {
+ t.Fatalf("purge POST: %v", err)
+ }
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("purge status = %d, want 200", resp.StatusCode)
+ }
+
+ if _, ok := cache.Get(CacheKey{Host: "a.example", Method: "GET", Path: "/event"}); ok {
+ t.Fatal("expected entry to be purged")
+ }
+}
+
+func TestListenPurgeCannotAffectAnotherHost(t *testing.T) {
+ dir := t.TempDir()
+ sock := filepath.Join(dir, "instance.sock")
+ cache := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
+ cache.Put(CacheKey{Host: "other.example", Method: "GET", Path: "/event"}, &CachedResponse{
+ StatusCode: 200, Header: http.Header{}, Body: []byte("hi"), StoredAt: time.Now(), TTL: time.Minute,
+ })
+
+ pl, err := ListenPurge(sock, "a.example", cache) // bound to a.example, not other.example
+ if err != nil {
+ t.Fatalf("ListenPurge: %v", err)
+ }
+ defer pl.Close()
+
+ body, _ := json.Marshal(map[string]string{"path": "/event", "query": ""})
+ client := dialPurgeSocket(t, sock)
+ resp, _ := client.Post("http://unix/purge", "application/json", bytes.NewReader(body))
+ resp.Body.Close()
+
+ if _, ok := cache.Get(CacheKey{Host: "other.example", Method: "GET", Path: "/event"}); !ok {
+ t.Fatal("a purge listener bound to a.example must never affect other.example's cache entries")
+ }
+}
+
+func TestListenPurgeSocketFileIsOwnerOnly(t *testing.T) {
+ dir := t.TempDir()
+ sock := filepath.Join(dir, "instance.sock")
+ pl, err := ListenPurge(sock, "a.example", NewCache(10, 1<<20))
+ if err != nil {
+ t.Fatalf("ListenPurge: %v", err)
+ }
+ defer pl.Close()
+
+ fi, err := os.Stat(sock)
+ if err != nil {
+ t.Fatalf("stat socket: %v", err)
+ }
+ if mode := fi.Mode().Perm(); mode != 0600 {
+ t.Fatalf("socket mode = %o, want 0600", mode)
+ }
+}
+
+func TestListenPurgeRemovesStaleSocketFile(t *testing.T) {
+ dir := t.TempDir()
+ sock := filepath.Join(dir, "instance.sock")
+ if err := os.WriteFile(sock, []byte("stale"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ pl, err := ListenPurge(sock, "a.example", NewCache(10, 1<<20))
+ if err != nil {
+ t.Fatalf("ListenPurge should clean up a stale socket file, got: %v", err)
+ }
+ pl.Close()
+}