diff --git a/README.md b/README.md index ccbbdf4..2f5558f 100644 --- a/README.md +++ b/README.md @@ -247,6 +247,59 @@ Three forms are accepted: `tcp://host:port`, `http://...` and `https://...`. > Readiness is polled only while the machine is live. A machine that sleeps all > day therefore generates no per-route traffic. +## Static response cache + +```toml +[cache] +enabled = true +root_path = "./cache" +ttl = "24h" +max_response_size_bytes = 10485760 +warm_interval = "30m" + +[[cache.routes]] +route = "files.home.com" +paths = ["/docs", "/assets/*"] +``` + +Some routes serve content that does not need the machine to be awake: a model +list, static assets, a landing page. With the cache enabled, a `GET` whose path +matches a configured pattern is served straight from disk when the machine is +offline, instead of waking it. The client gets a full answer while the box +stays asleep. + +A `/*` suffix caches everything under a prefix (`/assets/*` covers +`/assets/app.css`). Only `GET` requests with a 2xx status are cached, and only +for routes named in `[[cache.routes]]`; responses larger than +`max_response_size_bytes` are skipped. The query string is part of the identity +of an entry, so `/models?filter=a` is stored and replayed separately from +`/models`. + +Nothing that belongs to one client is ever stored, and nothing stored +anonymously is ever handed to an authenticated one: a request carrying an +`Authorization` or `Cookie` header is not cached and is not answered from cache +either, and neither is a response that sets a `Set-Cookie` or declares +`Cache-Control: private`, `no-store` or `no-cache`. A `Set-Cookie` header is +also dropped from anything that does get stored, so a cached page cannot hand +one visitor's session to the next. + +The cache answers only for a machine the health checker has confirmed down. A +machine that is up but whose health entry has gone stale still serves its own +response. + +The cache keeps itself warm: whenever a machine wakes up, and every +`warm_interval` while it stays up, the configured paths are re-fetched. A +response whose body has not changed is not rewritten, so a static file list +does not churn the disk. Entries older than `ttl` are dropped. Both `ttl` and +`warm_interval` must be positive - a zero or negative value is rejected as a +configuration error rather than expiring every entry or crashing the proxy at +startup. + +> [!NOTE] +> Cached responses are replayed with the upstream headers they were stored +> with, minus framing headers. The proxy does not revalidate them against the +> machine, so pick a `ttl` that matches how stale the content may get. + ## Idle shutdown ```toml diff --git a/cache_safety_test.go b/cache_safety_test.go new file mode 100644 index 0000000..2baeeaa --- /dev/null +++ b/cache_safety_test.go @@ -0,0 +1,527 @@ +package main + +import ( + "bytes" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// --- what may be cached ---------------------------------------------------- + +// TestCacheSkipsRequestsWithCredentials covers the leak where a logged-in +// client's response is stored and then replayed to every other client that asks +// for the same path while the machine is asleep. +func TestCacheSkipsRequestsWithCredentials(t *testing.T) { + backend := newCacheTestBackend(t, false) + dir := t.TempDir() + p, cs, _, _, _, _ := setupCacheProxy(t, backend, dir, []string{"/models"}) + + for _, header := range [][2]string{ + {"Authorization", "Bearer super-secret"}, + {"Cookie", "session=abc123"}, + } { + req := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + req.Header.Set(header[0], header[1]) + rec := httptest.NewRecorder() + p.handleRequest(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s request: status = %d, want 200", header[0], rec.Code) + } + } + + // The capture runs when the proxy closes the body, so give it a moment + // before insisting nothing landed. + time.Sleep(300 * time.Millisecond) + if _, ok := cs.Get("backend.test", "/models"); ok { + t.Fatal("a response fetched with credentials must not be cached") + } + if files := listCacheFiles(t, dir); len(files) != 0 { + t.Fatalf("cache files written for credentialed requests: %v", files) + } +} + +// TestCacheSkipsPrivateResponses covers the other half of the same leak: the +// response itself says it is not shareable. +func TestCacheSkipsPrivateResponses(t *testing.T) { + cases := []struct { + name string + headers http.Header + }{ + {"Set-Cookie", http.Header{"Set-Cookie": []string{"session=abc123; Path=/"}}}, + {"Cache-Control private", http.Header{"Cache-Control": []string{"private"}}}, + {"Cache-Control no-store", http.Header{"Cache-Control": []string{"no-store"}}}, + {"Cache-Control no-cache", http.Header{"Cache-Control": []string{"no-cache, max-age=0"}}}, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/health" { + w.WriteHeader(http.StatusOK) + return + } + for key, values := range tc.headers { + for _, value := range values { + w.Header().Add(key, value) + } + } + io.WriteString(w, `{"data":[{"id":"model-a"}]}`) + })) + defer backend.Close() + + p, cs, _, _, _, _ := setupCacheProxy(t, backend, t.TempDir(), []string{"/models"}) + + req := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec := httptest.NewRecorder() + p.handleRequest(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + + time.Sleep(300 * time.Millisecond) + if _, ok := cs.Get("backend.test", "/models"); ok { + t.Fatalf("a %s response must not be cached", tc.name) + } + }) + } +} + +// TestSanitizeCachedHeadersDropsSetCookie checks the second guard: even if a +// Set-Cookie response reached Save, the header is not stored with the entry. +func TestSanitizeCachedHeadersDropsSetCookie(t *testing.T) { + in := http.Header{ + "Set-Cookie": []string{"session=abc123; Path=/"}, + "Content-Type": []string{"application/json"}, + } + out := sanitizeCachedHeaders(in) + + if got := out.Values("Set-Cookie"); len(got) != 0 { + t.Errorf("Set-Cookie must be dropped, got %v", got) + } + if got := out.Get("Content-Type"); got != "application/json" { + t.Errorf("Content-Type = %q, want it preserved", got) + } +} + +// --- capture completeness -------------------------------------------------- + +// truncatedReader hands out its chunks and then fails with a non-EOF error, +// standing in for an upstream reset or a client that hung up mid-copy. +type truncatedReader struct { + chunks []string + next int +} + +// Read yields the next chunk, or an error once they are exhausted. +func (r *truncatedReader) Read(p []byte) (int, error) { + if r.next >= len(r.chunks) { + return 0, errors.New("upstream reset") + } + n := copy(p, r.chunks[r.next]) + r.next++ + return n, nil +} + +// TestCatchingBodySavesOnlyCompleteBodies pins the three capture rules: a body +// that ran to EOF is saved, one cut short by a read error is not, and one past +// the size limit is neither saved nor held in memory. +func TestCatchingBodySavesOnlyCompleteBodies(t *testing.T) { + t.Run("complete body is saved", func(t *testing.T) { + var saved []byte + cb := &catchingBody{ + original: io.NopCloser(strings.NewReader("hello world")), + buf: &bytes.Buffer{}, + limit: 64, + onDone: func(body []byte) { saved = body }, + } + if _, err := io.ReadAll(cb); err != nil { + t.Fatalf("read: %v", err) + } + cb.Close() + + if string(saved) != "hello world" { + t.Fatalf("saved = %q, want the whole body", saved) + } + }) + + t.Run("body cut short by a read error is not saved", func(t *testing.T) { + var saved []byte + cb := &catchingBody{ + original: io.NopCloser(&truncatedReader{chunks: []string{"hel"}}), + buf: &bytes.Buffer{}, + limit: 64, + onDone: func(body []byte) { saved = body }, + } + if _, err := io.ReadAll(cb); err == nil { + t.Fatal("the read error must reach the caller") + } + cb.Close() + + if saved != nil { + t.Fatalf("partial body %q must not be saved", saved) + } + }) + + t.Run("oversized body is not saved and stops being buffered", func(t *testing.T) { + var saved []byte + const limit = 8 + cb := &catchingBody{ + original: io.NopCloser(strings.NewReader(strings.Repeat("a", 4096))), + buf: &bytes.Buffer{}, + limit: limit, + onDone: func(body []byte) { saved = body }, + } + if _, err := io.ReadAll(cb); err != nil { + t.Fatalf("read: %v", err) + } + cb.Close() + + if saved != nil { + t.Fatalf("a body over the limit (%d bytes) must not be saved", len(saved)) + } + if !cb.overflow { + t.Error("overflow must be flagged once the limit is passed") + } + if cb.buf != nil { + t.Error("the buffer must be dropped once the body is over the limit") + } + }) +} + +// TestWarmMachineSkipsOversizedBody: io.LimitReader stops at the limit without +// an error, so reading exactly CacheMaxResponseSize bytes used to store a +// truncated body as a perfectly good 200. +func TestWarmMachineSkipsOversizedBody(t *testing.T) { + body := strings.Repeat("x", 32) + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/health" { + w.WriteHeader(http.StatusOK) + return + } + io.WriteString(w, body) + })) + defer backend.Close() + + dir := t.TempDir() + p, cs, _, _, _, machine := setupCacheProxy(t, backend, dir, []string{"/models"}) + cs.maxSize = 16 + p.config.CacheMaxResponseSize = 16 + + p.warmMachine(machine) + + if _, ok := cs.Get("backend.test", "/models"); ok { + t.Fatal("an oversized warm-up response must not be cached at all") + } + if files := listCacheFiles(t, dir); len(files) != 0 { + t.Fatalf("oversized warm-up wrote cache files: %v", files) + } +} + +// --- cache configuration --------------------------------------------------- + +// TestLoadConfigCacheRejectsNonPositiveDurations: warm_interval = "0s" reached +// time.NewTicker, which panics inside the warming goroutine and takes the whole +// proxy down at startup; a non-positive ttl expires every entry as it is saved. +func TestLoadConfigCacheRejectsNonPositiveDurations(t *testing.T) { + load := func(t *testing.T, ttl, warmInterval string) error { + t.Helper() + src := ` +port = ":9100" +timeout = "1m" +poll_interval = "5s" +health_check_interval = "30s" +health_cache_duration = "10s" + +[cache] +enabled = true +root_path = "` + filepath.Join(t.TempDir(), "cache") + `" +ttl = "` + ttl + `" +warm_interval = "` + warmInterval + `" + +[[cache.routes]] +route = "llamacpp.home.test" +paths = ["/models"] + +[[machines]] +name = "llamacpp" +mac_address = "A8:A1:59:40:D6:7D" +broadcast_ip = "192.168.50.255" +health_check = "http://192.168.50.2:8080/health" + +[[routes]] +machine = "llamacpp" +hostname = "llamacpp.home.test" +destination = "http://192.168.50.2:8080" +` + _, err := LoadConfig(writeTempConfig(t, src), RealClock{}) + return err + } + + for _, tc := range []struct{ ttl, warmInterval, want string }{ + {"0s", "10m", "cache ttl"}, + {"-1h", "10m", "cache ttl"}, + {"24h", "0s", "warm_interval"}, + {"24h", "-5m", "warm_interval"}, + } { + err := load(t, tc.ttl, tc.warmInterval) + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Errorf("ttl=%q warm_interval=%q: got %v, want an error mentioning %q", + tc.ttl, tc.warmInterval, err, tc.want) + } + } + + if err := load(t, "24h", "10m"); err != nil { + t.Errorf("positive durations must load, got: %v", err) + } +} + +// --- who gets served from cache -------------------------------------------- + +// TestCacheNotServedForHealthyMachineWithStaleHealth: the health cache going +// stale is not evidence the machine is down. Only a machine the health checker +// confirmed unhealthy may be answered from disk. +func TestCacheNotServedForHealthyMachineWithStaleHealth(t *testing.T) { + var served int32 + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/health" { + w.WriteHeader(http.StatusOK) + return + } + n := atomic.AddInt32(&served, 1) + fmt.Fprintf(w, `{"data":[{"id":"live-%d"}]}`, n) + })) + defer backend.Close() + + p, cs, _, hc, _, machine := setupCacheProxy(t, backend, t.TempDir(), []string{"/models"}) + + req := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec := httptest.NewRecorder() + p.handleRequest(rec, req) + waitFor(t, 3*time.Second, "cache entry after the first proxied GET", func() bool { + _, ok := cs.Get("backend.test", "/models") + return ok + }) + + // Healthy, but with an entry old enough that healthCacheStatus calls it + // stale. The machine is up: it must answer in person. + hc.setResult(true) + machine.mu.Lock() + machine.IsHealthy = true + machine.LastCheck = time.Now().Add(-time.Hour) + machine.mu.Unlock() + + req2 := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec2 := httptest.NewRecorder() + p.handleRequest(rec2, req2) + + body, _ := io.ReadAll(rec2.Result().Body) + if rec2.Code != http.StatusOK { + t.Fatalf("status = %d, want 200 from the live machine", rec2.Code) + } + if !strings.Contains(string(body), "live-2") { + t.Fatalf("got %q, want the live response; a stale health entry must not serve the cache", body) + } +} + +// TestCacheKeyIncludesQueryString: /models?filter=a and /models are different +// responses, so whichever was fetched last must not answer for the other. +func TestCacheKeyIncludesQueryString(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/health" { + w.WriteHeader(http.StatusOK) + return + } + fmt.Fprintf(w, `{"query":%q}`, r.URL.RawQuery) + })) + defer backend.Close() + + p, cs, wol, hc, _, machine := setupCacheProxy(t, backend, t.TempDir(), []string{"/models"}) + + req := httptest.NewRequest("GET", "http://backend.test:9100/models?filter=a", nil) + rec := httptest.NewRecorder() + p.handleRequest(rec, req) + waitFor(t, 3*time.Second, "cache entry for the filtered path", func() bool { + _, ok := cs.Get("backend.test", "/models?filter=a") + return ok + }) + + hc.setResult(false) + setMachineHealth(machine, false) + p.config.Timeout = 1 * time.Second + + // The unfiltered path was never fetched: the cache must not answer for it. + req2 := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec2 := httptest.NewRecorder() + p.handleRequest(rec2, req2) + if rec2.Code != http.StatusServiceUnavailable { + t.Fatalf("GET /models while down: status=%d, want 503 rather than the filtered response", rec2.Code) + } + if wol.callCount() == 0 { + t.Error("a cache miss must still attempt a wake") + } + + // The exact URL that was fetched is served from cache. + req3 := httptest.NewRequest("GET", "http://backend.test:9100/models?filter=a", nil) + rec3 := httptest.NewRecorder() + p.handleRequest(rec3, req3) + body, _ := io.ReadAll(rec3.Result().Body) + if rec3.Code != http.StatusOK || !strings.Contains(string(body), `filter=a`) { + t.Fatalf("GET /models?filter=a while down: status=%d body=%q", rec3.Code, body) + } +} + +// --- on-disk writes -------------------------------------------------------- + +// TestWriteFileAtomicConcurrentWriters: the wake-triggered warm-up, the +// periodic warm-up and a proxied GET can all save the same key at once. A +// shared temp path let them interleave into a corrupt envelope before the +// rename, so each write gets its own temp file. +func TestWriteFileAtomicConcurrentWriters(t *testing.T) { + dir := t.TempDir() + filePath := filepath.Join(dir, "entry.cache") + + const writers = 8 + payload := func(i int) []byte { return []byte(strings.Repeat(string(rune('A'+i)), 4096)) } + + var wg sync.WaitGroup + for i := 0; i < writers; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + if err := writeFileAtomic(filePath, payload(i)); err != nil { + t.Errorf("writeFileAtomic: %v", err) + } + }(i) + } + wg.Wait() + + data, err := os.ReadFile(filePath) + if err != nil { + t.Fatalf("published entry: %v", err) + } + whole := false + for i := 0; i < writers; i++ { + if bytes.Equal(data, payload(i)) { + whole = true + break + } + } + if !whole { + t.Fatalf("published entry is a mix of writers: %d bytes, first byte %q", len(data), data[:1]) + } + + leftovers, _ := filepath.Glob(filepath.Join(dir, "*.tmp")) + if len(leftovers) != 0 { + t.Errorf("temp files left behind: %v", leftovers) + } +} + +// TestWarmMachineSkipsUnshareableResponse: the warm-up client sends no +// credentials, but the backend can still answer with a cookie or a +// Cache-Control that says the body is not shareable. Save strips Set-Cookie and +// keeps the body, so the guard has to be in the warm-up path too - otherwise a +// later offline GET is handed a response the upstream never meant to give it. +func TestWarmMachineSkipsUnshareableResponse(t *testing.T) { + for _, tc := range []struct{ name, header, value string }{ + {"Set-Cookie", "Set-Cookie", "session=abc123; Path=/"}, + {"Cache-Control private", "Cache-Control", "private"}, + {"Cache-Control no-store", "Cache-Control", "no-store"}, + } { + tc := tc + t.Run(tc.name, func(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/health" { + w.WriteHeader(http.StatusOK) + return + } + w.Header().Set(tc.header, tc.value) + io.WriteString(w, `{"data":[{"id":"model-a"}]}`) + })) + defer backend.Close() + + p, cs, _, _, _, machine := setupCacheProxy(t, backend, t.TempDir(), []string{"/models"}) + p.warmMachine(machine) + + if _, ok := cs.Get("backend.test", "/models"); ok { + t.Fatalf("a warmed %s response must not be cached", tc.name) + } + }) + } +} + +// TestCacheNotServedForCredentialedRequest: every stored entry was fetched +// anonymously, so replaying it to a logged-in client answers that client with a +// response the backend never produced for it. +func TestCacheNotServedForCredentialedRequest(t *testing.T) { + backend := newCacheTestBackend(t, false) + p, cs, wol, hc, _, machine := setupCacheProxy(t, backend, t.TempDir(), []string{"/models"}) + + req := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec := httptest.NewRecorder() + p.handleRequest(rec, req) + waitFor(t, 3*time.Second, "cache entry after an anonymous GET", func() bool { + _, ok := cs.Get("backend.test", "/models") + return ok + }) + + hc.setResult(false) + setMachineHealth(machine, false) + p.config.Timeout = 1 * time.Second + + req2 := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + req2.Header.Set("Cookie", "session=abc123") + rec2 := httptest.NewRecorder() + p.handleRequest(rec2, req2) + + if rec2.Code == http.StatusOK { + t.Fatal("a credentialed request must not be answered with the anonymous cached response") + } + if rec2.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want 503 while the machine is down", rec2.Code) + } + if wol.callCount() == 0 { + t.Error("a credentialed cache miss must still attempt a wake") + } +} + +// TestCacheSaveRenewsTimestampForUnchangedContent: warm-up re-fetches a body +// that never changes, and Save skips the write when the bytes match. That left +// the Timestamp inside the envelope at the original save time, so an entry that +// was being refreshed every cycle still expired and stopped being served. +func TestCacheSaveRenewsTimestampForUnchangedContent(t *testing.T) { + const ttl = time.Second + cs := NewCacheStore(filepath.Join(t.TempDir(), "cache"), ttl, 1<<20, + map[string][]string{"backend.test": {"/models"}}, noopLogger{}) + defer cs.Close() + + headers := http.Header{"Content-Type": []string{"application/json"}} + body := []byte(`{"data":[{"id":"model-a"}]}`) + + cs.Save("backend.test", "/models", http.StatusOK, headers, body) + if _, ok := cs.Get("backend.test", "/models"); !ok { + t.Fatal("a freshly saved entry must be readable") + } + + // Past half the TTL: the next identical save has to renew the envelope + // instead of skipping the write. + time.Sleep(700 * time.Millisecond) + cs.Save("backend.test", "/models", http.StatusOK, headers, body) + + // From here the original save time has aged out of the TTL, so only a + // renewed envelope is still servable. + time.Sleep(500 * time.Millisecond) + if _, ok := cs.Get("backend.test", "/models"); !ok { + t.Fatal("an entry warm-up keeps refreshing must stay inside its TTL") + } +} diff --git a/cache_test.go b/cache_test.go new file mode 100644 index 0000000..acbdd55 --- /dev/null +++ b/cache_test.go @@ -0,0 +1,639 @@ +package main + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path" + "path/filepath" + "strings" + "testing" + "time" +) + +// --- helpers --------------------------------------------------------------- + +func newCacheTestBackend(t *testing.T, chunked bool) *httptest.Server { + t.Helper() + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/health": + w.WriteHeader(http.StatusOK) + case "/models", "/v1/models": + w.Header().Set("Content-Type", "application/json") + if chunked { + // Force a chunked response (no Content-Length) by flushing + // partway through the body. + if fl, ok := w.(http.Flusher); ok { + io.WriteString(w, `{"data":[{"id":"model-`) + fl.Flush() + io.WriteString(w, `a"}]}`) + return + } + } + io.WriteString(w, `{"data":[{"id":"model-a"}]}`) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(backend.Close) + return backend +} + +// setupCacheProxy wires a proxy with one machine ("llamacpp"), one HTTP route +// named after its hostname, and a cache rooted at cacheDir caching paths for +// that route. +func setupCacheProxy(t *testing.T, backend *httptest.Server, cacheDir string, paths []string) (*ProxyService, *CacheStore, *mockWOLSender, *mockHealthChecker, *recordingLogger, *Machine) { + t.Helper() + logger := &recordingLogger{} + + routeName := "backend.test" + cs := NewCacheStore(cacheDir, 240*time.Hour, 10<<20, map[string][]string{routeName: paths}, logger) + + machine := &Machine{ + Name: "llamacpp", + Config: &MachineConfig{ + MacAddress: "A8:A1:59:40:D6:7D", + BroadcastIP: "127.0.0.1", + WolPort: 9, + HealthCheck: backend.URL + "/health", + }, + IsHealthy: true, + LastCheck: time.Now(), + LastActivity: time.Now(), + } + + route := &Route{ + Name: routeName, + Machine: machine, + Hostname: "backend.test", + Destination: backend.URL, + HealthCheck: backend.URL + "/health", + IsReady: true, + LastCheck: time.Now(), + } + + cfg := &ProxyConfig{ + Port: "127.0.0.1:0", + Timeout: 10 * time.Second, + PollInterval: 1 * time.Second, + ResponseHeaderTimeout: time.Minute, + HealthCheckInterval: 30 * time.Second, + HealthCacheDuration: 10 * time.Second, + Machines: map[string]*Machine{"llamacpp": machine}, + Routes: []*Route{route}, + RoutesByHostname: map[string]*Route{"backend.test": route}, + CacheEnabled: true, + CacheRootPath: cacheDir, + CacheTTL: 240 * time.Hour, + CacheMaxResponseSize: 10 << 20, + CacheWarmInterval: 10 * time.Minute, + CacheRoutes: map[string][]string{routeName: paths}, + } + + wol := newMockWOLSender() + hc := &mockHealthChecker{} + hc.setResult(true) + p := NewProxyService(cfg, hc, wol, newMockSSHExecutor(), logger, RealClock{}, cs) + return p, cs, wol, hc, logger, machine +} + +// setMachineHealth records a health result on the machine under its lock, the +// way the health checker would. +func setMachineHealth(machine *Machine, healthy bool) { + machine.mu.Lock() + machine.IsHealthy = healthy + machine.LastCheck = time.Now() + machine.mu.Unlock() +} + +// listCacheFiles names the entry files the cache has written for the test +// route, so a test can assert nothing was stored rather than merely unreadable. +func listCacheFiles(t *testing.T, dir string) []string { + t.Helper() + var out []string + entries, err := os.ReadDir(filepath.Join(dir, "backend.test")) + if err != nil { + return out + } + for _, e := range entries { + out = append(out, e.Name()) + } + return out +} + +// --- config ---------------------------------------------------------------- + +func TestLoadConfigCache(t *testing.T) { + const toml = ` +port = ":9100" +timeout = "1m" +poll_interval = "5s" +response_header_timeout = "10m" +health_check_interval = "30s" +health_cache_duration = "10s" + +[cache] +enabled = true +root_path = "/app/cache" +ttl = "240h" +max_response_size_bytes = 10485760 +warm_interval = "10m" + +[[cache.routes]] +route = "llamacpp.home.test" +paths = ["/v1/models", "/models"] + +[[machines]] +name = "llamacpp" +mac_address = "A8:A1:59:40:D6:7D" +broadcast_ip = "192.168.50.255" +wol_port = 9 +health_check = "http://192.168.50.2:8080/health" + +[[routes]] +machine = "llamacpp" +hostname = "llamacpp.home.test" +destination = "http://192.168.50.2:8080" +` + path := filepath.Join(t.TempDir(), "config.toml") + if err := os.WriteFile(path, []byte(toml), 0644); err != nil { + t.Fatal(err) + } + + cfg, err := LoadConfig(path, RealClock{}) + if err != nil { + t.Fatalf("config failed to load: %v", err) + } + if !cfg.CacheEnabled { + t.Error("cache should be enabled") + } + if got := cfg.CacheRootPath; got != "/app/cache" { + t.Errorf("CacheRootPath = %q, want /app/cache", got) + } + if got := cfg.CacheTTL; got != 240*time.Hour { + t.Errorf("CacheTTL = %v, want 240h", got) + } + if got := cfg.CacheMaxResponseSize; got != 10485760 { + t.Errorf("CacheMaxResponseSize = %d, want 10485760", got) + } + if got := cfg.CacheWarmInterval; got != 10*time.Minute { + t.Errorf("CacheWarmInterval = %v, want 10m", got) + } + paths := cfg.CacheRoutes["llamacpp.home.test"] + if len(paths) != 2 || paths[0] != "/v1/models" || paths[1] != "/models" { + t.Fatalf("CacheRoutes[llamacpp.home.test] = %v, want [/v1/models /models]", paths) + } +} + +// TestLoadConfigCacheDefaults checks that an enabled cache with only +// enabled=true and a route entry gets the documented defaults. +func TestLoadConfigCacheDefaults(t *testing.T) { + const toml = ` +port = ":9100" +timeout = "1m" +poll_interval = "5s" +health_check_interval = "30s" +health_cache_duration = "10s" + +[cache] +enabled = true + +[[machines]] +name = "nas" +mac_address = "AA:BB:CC:DD:EE:FF" +broadcast_ip = "10.0.0.255" +health_check = "tcp://nas.local:22" + +[[routes]] +machine = "nas" +hostname = "nas.home.test" +destination = "http://nas.local" +` + path := filepath.Join(t.TempDir(), "config.toml") + if err := os.WriteFile(path, []byte(toml), 0644); err != nil { + t.Fatal(err) + } + + cfg, err := LoadConfig(path, RealClock{}) + if err != nil { + t.Fatalf("config failed to load: %v", err) + } + if !cfg.CacheEnabled { + t.Fatal("cache should be enabled") + } + if got := cfg.CacheRootPath; got != "./cache" { + t.Errorf("CacheRootPath = %q, want ./cache", got) + } + if got := cfg.CacheTTL; got != 24*time.Hour { + t.Errorf("CacheTTL = %v, want 24h", got) + } + if got := cfg.CacheMaxResponseSize; got != 10*1024*1024 { + t.Errorf("CacheMaxResponseSize = %d, want 10485760", got) + } + if got := cfg.CacheWarmInterval; got != 30*time.Minute { + t.Errorf("CacheWarmInterval = %v, want 30m", got) + } +} + +// TestLoadConfigCacheRejectsUnknownRoute checks that a cache entry naming a +// route the config does not define is a configuration error rather than a +// silently dead cache. +func TestLoadConfigCacheRejectsUnknownRoute(t *testing.T) { + const toml = ` +port = ":9100" +timeout = "1m" +poll_interval = "5s" +health_check_interval = "30s" +health_cache_duration = "10s" + +[cache] +enabled = true + +[[cache.routes]] +route = "typo.home.test" +paths = ["/models"] + +[[machines]] +name = "nas" +mac_address = "AA:BB:CC:DD:EE:FF" +broadcast_ip = "10.0.0.255" +health_check = "tcp://nas.local:22" + +[[routes]] +machine = "nas" +hostname = "nas.home.test" +destination = "http://nas.local" +` + path := filepath.Join(t.TempDir(), "config.toml") + if err := os.WriteFile(path, []byte(toml), 0644); err != nil { + t.Fatal(err) + } + + if _, err := LoadConfig(path, RealClock{}); err == nil || + !strings.Contains(err.Error(), `unknown route "typo.home.test"`) { + t.Errorf("cache naming a non-existent route must be rejected, got: %v", err) + } +} + +// TestLoadConfigCacheRejectsTCPRoute checks that a cache entry for a TCP route +// is rejected: the cache stores HTTP responses and cannot answer a raw socket. +func TestLoadConfigCacheRejectsTCPRoute(t *testing.T) { + const toml = ` +port = ":9100" +timeout = "1m" +poll_interval = "5s" +health_check_interval = "30s" +health_cache_duration = "10s" + +[cache] +enabled = true + +[[cache.routes]] +route = ":2222" +paths = ["/models"] + +[[machines]] +name = "nas" +mac_address = "AA:BB:CC:DD:EE:FF" +broadcast_ip = "10.0.0.255" +health_check = "tcp://nas.local:22" + +[[routes]] +machine = "nas" +listen_port = 2222 +destination = "nas.local:22" +` + path := filepath.Join(t.TempDir(), "config.toml") + if err := os.WriteFile(path, []byte(toml), 0644); err != nil { + t.Fatal(err) + } + + if _, err := LoadConfig(path, RealClock{}); err == nil || + !strings.Contains(err.Error(), "only HTTP routes can be cached") { + t.Errorf("cache naming a TCP route must be rejected, got: %v", err) + } +} + +// --- serving --------------------------------------------------------------- + +// Proxying a GET /models while the machine is healthy must populate the cache +// on disk (ModifyResponse body capture -> Save). Once the machine is down the +// cached response must be served without sending a single WOL packet. +func TestCachePopulatedViaProxiedResponse(t *testing.T) { + backend := newCacheTestBackend(t, false) + p, cs, wol, hc, _, machine := setupCacheProxy(t, backend, t.TempDir(), []string{"/models"}) + + req := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec := httptest.NewRecorder() + p.handleRequest(rec, req) + + if rec.Code != 200 { + t.Fatalf("expected 200, got %d", rec.Code) + } + // The proxied body is cached when the proxy closes it, which can lag the + // recorder by a fraction of a second: poll briefly. + waitFor(t, 3*time.Second, "cache entry after proxied GET", func() bool { + _, ok := cs.Get("backend.test", "/models") + return ok + }) + + // Machine goes down: the cached response must be served, no WOL. + hc.setResult(false) + setMachineHealth(machine, false) + + req2 := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec2 := httptest.NewRecorder() + p.handleRequest(rec2, req2) + + body, _ := io.ReadAll(rec2.Result().Body) + if rec2.Code != 200 || !strings.Contains(string(body), "model-a") { + t.Fatalf("cached response wrong: status=%d body=%q", rec2.Code, body) + } + if got := wol.callCount(); got != 0 { + t.Errorf("WOL sent %d time(s) although the cache could answer", got) + } +} + +// A chunked upstream response (no Content-Length, like many small JSON +// servers send) must be cached and replayed in full. +func TestCachedChunkedResponseServedCorrectly(t *testing.T) { + backend := newCacheTestBackend(t, true) + p, cs, _, hc, _, machine := setupCacheProxy(t, backend, t.TempDir(), []string{"/models"}) + + // Populate while healthy. + req := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec := httptest.NewRecorder() + p.handleRequest(rec, req) + if rec.Code != 200 { + t.Fatalf("expected 200, got %d", rec.Code) + } + waitFor(t, 3*time.Second, "chunked response cached", func() bool { + _, ok := cs.Get("backend.test", "/models") + return ok + }) + + // Serve from cache while down. + hc.setResult(false) + setMachineHealth(machine, false) + + req2 := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec2 := httptest.NewRecorder() + p.handleRequest(rec2, req2) + + body, _ := io.ReadAll(rec2.Result().Body) + if rec2.Code != 200 || string(body) != `{"data":[{"id":"model-a"}]}` { + t.Fatalf("served chunked-cache response wrong: status=%d body=%q", rec2.Code, body) + } +} + +// warmMachine must (re)populate the cache for all configured paths of every +// HTTP route pointing at the machine. +func TestWarmMachinePopulates(t *testing.T) { + backend := newCacheTestBackend(t, false) + p, cs, _, _, _, machine := setupCacheProxy(t, backend, t.TempDir(), []string{"/v1/models", "/models"}) + + p.warmMachine(machine) + + files := listCacheFiles(t, cs.rootPath) + t.Logf("cache files after warm: %v", files) + // Two paths -> one single-file entry each. + if len(files) != 2 { + t.Fatalf("expected 2 cache entries, got %v", files) + } + if _, ok := cs.Get("backend.test", "/models"); !ok { + t.Error("/models not cached after warm") + } + if _, ok := cs.Get("backend.test", "/v1/models"); !ok { + t.Error("/v1/models not cached after warm") + } +} + +// Entries written by older versions (body file + ".meta" sidecar) must keep +// working after the upgrade, so an upgraded container does not start waking +// the machine for responses that are already cached on disk. +func TestLegacyCacheEntriesStillLoadable(t *testing.T) { + backend := newCacheTestBackend(t, false) + cacheDir := t.TempDir() + logger := &recordingLogger{} + + // Seed a legacy two-file entry before the store starts so that loadAll + // has to discover it. The key must be computed the same way the + // production code does: from the path.Clean-ed path. + { + probe := NewCacheStore(cacheDir, 240*time.Hour, 10<<20, map[string][]string{"backend.test": {"/models"}}, logger) + cleaned := path.Clean("/models") + k := probe.key("backend.test", cleaned) + dir := filepath.Join(cacheDir, "backend.test") + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatal(err) + } + body := []byte(`{"data":[{"id":"legacy-model"}]}`) + if err := os.WriteFile(filepath.Join(dir, k+".cache"), body, 0644); err != nil { + t.Fatal(err) + } + pathJSON, err := json.Marshal(cleaned) + if err != nil { + t.Fatal(err) + } + meta := `{"status_code":200,"headers":{"Content-Type":["application/json"]},"timestamp":"` + + time.Now().Add(-time.Hour).Format(time.RFC3339) + `","path":` + string(pathJSON) + `}` + if err := os.WriteFile(filepath.Join(dir, k+".cache.meta"), []byte(meta), 0644); err != nil { + t.Fatal(err) + } + } + + p, _, wol, hc, _, machine := setupCacheProxy(t, backend, cacheDir, []string{"/models"}) + + // Machine down: the legacy entry must be served from the cache, no WOL. + hc.setResult(false) + setMachineHealth(machine, false) + + req := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec := httptest.NewRecorder() + p.handleRequest(rec, req) + + body, _ := io.ReadAll(rec.Result().Body) + if rec.Code != 200 || !strings.Contains(string(body), "legacy-model") { + t.Fatalf("legacy entry not served: status=%d body=%q", rec.Code, body) + } + if got := wol.callCount(); got != 0 { + t.Errorf("WOL sent for a request a legacy cache entry could serve (sent=%d)", got) + } +} + +// --- CacheStore internals --------------------------------------------------- + +// sanitizeCachedHeaders must drop framing/hop-by-hop headers so a cached +// response is re-framed correctly when served from disk. +func TestSanitizeCachedHeaders(t *testing.T) { + h := http.Header{} + h.Set("Content-Type", "application/json") + h.Set("Content-Length", "42") + h.Set("Transfer-Encoding", "chunked") + h.Set("Connection", "keep-alive") + h.Set("X-Custom", "kept") + + out := sanitizeCachedHeaders(h) + for _, k := range []string{"Content-Length", "Transfer-Encoding", "Connection"} { + if _, ok := out[k]; ok { + t.Errorf("%s must be stripped, got %v", k, out[k]) + } + } + if out.Get("Content-Type") != "application/json" || out.Get("X-Custom") != "kept" { + t.Errorf("normal headers must be kept: %v", out) + } +} + +// Saving an unchanged body must skip the disk write while the entry is still +// on disk, but must write again once the file is gone: skipping the write on +// an unchanged body whose file vanished (TTL expiry, manual cleanup) would +// leave the cache empty forever for static responses, sending WOL on every +// offline request. +func TestCacheSaveSkipsUnchangedButRewritesMissingFile(t *testing.T) { + dir := t.TempDir() + cs := NewCacheStore(dir, 240*time.Hour, 10<<20, map[string][]string{"r": {"/p"}}, &recordingLogger{}) + + body := []byte("same") + cs.Save("r", "/p", 200, http.Header{"Content-Type": []string{"text/plain"}}, body) + + first, err := os.Stat(cs.cacheFilePath("r", "/p")) + if err != nil { + t.Fatal(err) + } + + cs.Save("r", "/p", 200, http.Header{"Content-Type": []string{"text/plain"}}, body) + second, err := os.Stat(cs.cacheFilePath("r", "/p")) + if err != nil { + t.Fatal(err) + } + if !first.ModTime().Equal(second.ModTime()) { + t.Error("unchanged body must not rewrite the cache file") + } + + if err := os.Remove(cs.cacheFilePath("r", "/p")); err != nil { + t.Fatal(err) + } + cs.Save("r", "/p", 200, http.Header{"Content-Type": []string{"text/plain"}}, body) + if _, ok := cs.Get("r", "/p"); !ok { + t.Error("unchanged body must be rewritten after the cache file disappeared") + } +} + +// Responses larger than the configured maximum must not be cached. +func TestCacheRejectsOversizedResponse(t *testing.T) { + dir := t.TempDir() + cs := NewCacheStore(dir, 240*time.Hour, 10, map[string][]string{"r": {"/p"}}, &recordingLogger{}) + + cs.Save("r", "/p", 200, http.Header{}, []byte(strings.Repeat("x", 11))) + if _, ok := cs.Get("r", "/p"); ok { + t.Error("oversized response must not be cached") + } +} + +// Cleanup must drop expired entries from memory and from disk. +func TestCacheCleanupRemovesExpired(t *testing.T) { + dir := t.TempDir() + cs := NewCacheStore(dir, time.Millisecond, 10<<20, map[string][]string{"r": {"/p"}}, &recordingLogger{}) + + cs.Save("r", "/p", 200, http.Header{}, []byte("body")) + time.Sleep(5 * time.Millisecond) + + cs.Cleanup() + + if _, ok := cs.Get("r", "/p"); ok { + t.Error("expired entry must be gone from memory") + } + if _, err := os.Stat(cs.cacheFilePath("r", "/p")); !os.IsNotExist(err) { + t.Error("expired entry must be removed from disk") + } +} + +// A "/*" pattern must warm every concrete path already cached under its +// prefix, while an exact pattern warms only itself. +func TestGetCachedPathsForRouteWildcard(t *testing.T) { + cs := NewCacheStore(t.TempDir(), 240*time.Hour, 10<<20, + map[string][]string{"r": {"/assets/*", "/docs"}}, &recordingLogger{}) + + cs.Save("r", "/assets/a.css", 200, http.Header{}, []byte("a")) + cs.Save("r", "/assets/b.js", 200, http.Header{}, []byte("b")) + cs.Save("r", "/docs", 200, http.Header{}, []byte("d")) + + got := cs.GetCachedPathsForRoute("r") + want := map[string]bool{"/docs": true, "/assets/a.css": true, "/assets/b.js": true} + if len(got) != len(want) { + t.Fatalf("GetCachedPathsForRoute = %v, want %v", got, want) + } + for _, p := range got { + if !want[p] { + t.Fatalf("GetCachedPathsForRoute = %v, want %v", got, want) + } + } +} + +// The cache must not answer non-GET requests: a cached body replayed for a +// POST would hide the write from the machine. +func TestCacheDoesNotServeNonGET(t *testing.T) { + backend := newCacheTestBackend(t, false) + p, cs, wol, hc, _, machine := setupCacheProxy(t, backend, t.TempDir(), []string{"/models"}) + + // Populate the cache via a healthy GET. + req := httptest.NewRequest("GET", "http://backend.test:9100/models", nil) + rec := httptest.NewRecorder() + p.handleRequest(rec, req) + waitFor(t, 3*time.Second, "cache entry after proxied GET", func() bool { + _, ok := cs.Get("backend.test", "/models") + return ok + }) + + hc.setResult(false) + setMachineHealth(machine, false) + + // Shorten the wake so the 503 arrives quickly: the point of the test is + // that the cache does not answer the POST, not the wake itself. + p.config.Timeout = 1 * time.Second + p.config.PollInterval = 200 * time.Millisecond + + req2 := httptest.NewRequest("POST", "http://backend.test:9100/models", strings.NewReader("{}")) + rec2 := httptest.NewRecorder() + p.handleRequest(rec2, req2) + + // The machine is down and the health check fails: the wake must time out + // with a 503 rather than being answered from the cache. + if rec2.Code != http.StatusServiceUnavailable { + t.Fatalf("POST while down: status=%d, want 503 (cache must not answer writes)", rec2.Code) + } + if wol.callCount() == 0 { + t.Error("POST while down must attempt a wake") + } +} + +// A cached response must be served with its stored headers and status, and +// must not leak the wake path: this is the exact scenario from the upstream +// issue - static assets served immediately while the target is asleep. +func TestCacheServePreservesStatusAndHeaders(t *testing.T) { + dir := t.TempDir() + cs := NewCacheStore(dir, 240*time.Hour, 10<<20, map[string][]string{"r": {"/p"}}, &recordingLogger{}) + cs.Save("r", "/p", 200, http.Header{"Content-Type": []string{"application/json"}, "Cache-Control": []string{"max-age=60"}}, []byte(`{"ok":true}`)) + + resp, ok := cs.Get("r", "/p") + if !ok { + t.Fatal("entry not found") + } + if resp.StatusCode != 200 { + t.Errorf("status = %d, want 200", resp.StatusCode) + } + if got := resp.Header.Get("Content-Type"); got != "application/json" { + t.Errorf("Content-Type = %q", got) + } + if got := resp.Header.Get("Cache-Control"); got != "max-age=60" { + t.Errorf("Cache-Control = %q", got) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if string(body) != `{"ok":true}` { + t.Errorf("body = %q", body) + } +} diff --git a/config.toml b/config.toml index 90722ec..d8382d8 100644 --- a/config.toml +++ b/config.toml @@ -19,6 +19,23 @@ poll_interval = "5s" # How often to check health while waiting health_check_interval = "30s" # Background health check frequency health_cache_duration = "10s" # How long to trust a cached health result +# Optional static response cache. When a machine is offline, GET responses +# for the configured paths are served from disk instead of waking it: the +# client gets a full answer while the box stays asleep. The cached paths are +# refreshed again whenever the machine wakes up. +[cache] +enabled = false # Enable static response caching +root_path = "./cache" # Directory to store cached responses +ttl = "24h" # How long cached entries remain valid. Must be positive. +max_response_size_bytes = 10485760 # Maximum response size to cache (10MB) +warm_interval = "30m" # How often to refresh cached paths while a machine is up. Must be positive. + +# Which paths to cache, per route. A "/*" suffix caches everything under a +# prefix (e.g. every file under /assets/). +# [[cache.routes]] +# route = "files.home.com" # Route name: its hostname +# paths = ["/docs", "/assets/*"] + # Optional SSL configuration. Do not add these values unless you plan to use TLS/HTTPS #ssl_certificate = "/path/to/cert.pem" #ssl_certificate_key = "/path/to/key.pem" diff --git a/doormouse.exe b/doormouse.exe new file mode 100644 index 0000000..ecaa4b3 Binary files /dev/null and b/doormouse.exe differ diff --git a/main.go b/main.go index 0bc73a8..da84c95 100644 --- a/main.go +++ b/main.go @@ -3,7 +3,10 @@ package main import ( "bytes" "context" + "crypto/sha256" "crypto/tls" + "encoding/binary" + "encoding/json" "errors" "fmt" "io" @@ -14,6 +17,7 @@ import ( "net/url" "os" "os/signal" + "path" "path/filepath" "strings" "sync" @@ -43,6 +47,7 @@ type SSHExecutor interface { type Logger interface { Info(msg string, args ...interface{}) Error(msg string, args ...interface{}) + Debug(msg string, args ...interface{}) } const tcpScheme = "tcp://" @@ -116,6 +121,7 @@ type Config struct { HealthCacheDuration string `toml:"health_cache_duration"` SSLCertificate string `toml:"ssl_certificate,omitempty"` SSLCertificateKey string `toml:"ssl_certificate_key,omitempty"` + CacheConfig CacheConfig `toml:"cache,omitempty"` Targets []Target `toml:"targets,omitempty"` // legacy, superseded by machines + routes Machines []MachineEntry `toml:"machines,omitempty"` Routes []RouteEntry `toml:"routes,omitempty"` @@ -147,6 +153,25 @@ type RouteEntry struct { HealthCheck string `toml:"health_check,omitempty"` } +// CacheConfig is the optional [cache] block. When a machine is offline, GET +// responses for the configured paths are served from disk instead of waking +// it: the client gets a full answer while the box stays asleep, and the +// cache is refreshed again once the machine is reachable. +type CacheConfig struct { + Enabled bool `toml:"enabled"` + RootPath string `toml:"root_path,omitempty"` + TTL string `toml:"ttl,omitempty"` + MaxResponseSizeBytes int64 `toml:"max_response_size_bytes,omitempty"` + WarmInterval string `toml:"warm_interval,omitempty"` + Routes []CacheRouteConfig `toml:"routes,omitempty"` +} + +// CacheRouteConfig is a [[cache.routes]] block: the paths cached for one route. +type CacheRouteConfig struct { + Route string `toml:"route"` + Paths []string `toml:"paths"` +} + type Target struct { Name string `toml:"name"` Hostname string `toml:"hostname"` @@ -228,6 +253,12 @@ type ProxyConfig struct { RoutesByListenPort map[int]*Route SSLCertificate string SSLCertificateKey string + CacheEnabled bool + CacheRootPath string + CacheTTL time.Duration + CacheMaxResponseSize int64 + CacheWarmInterval time.Duration + CacheRoutes map[string][]string // route name -> cached path patterns } type Machine struct { @@ -270,6 +301,604 @@ func (m *Machine) releaseConnection(now time.Time) { m.mu.Unlock() } +// --------------------------------------------------------------------------- +// Static response cache +// --------------------------------------------------------------------------- + +type cacheEntry struct { + StatusCode int `json:"status_code"` + Headers map[string][]string `json:"headers"` + Body []byte `json:"-"` + Timestamp time.Time `json:"timestamp"` + Path string `json:"path"` +} + +// CacheStore persists GET responses for configured route paths on disk so +// they can be served while the machine behind the route is asleep. +type CacheStore struct { + rootPath string + ttl time.Duration + maxSize int64 + logger Logger + entriesMu sync.RWMutex + entries map[string]*cacheEntry + routePaths map[string][]string // route name -> list of patterns + keyToRoute map[string]string // cache key -> route name + keyToPath map[string]string // cache key -> path + accessOrder []string + maxEntries int + lastHashes map[string][]byte // cache key -> last SHA-256 hash of body +} + +// NewCacheStore returns a cache rooted at rootPath, holding entries for ttl, +// rejecting bodies over maxSize, and covering the per-route path patterns in +// routePaths. It creates the root and loads whatever is already on disk. +func NewCacheStore(rootPath string, ttl time.Duration, maxSize int64, routePaths map[string][]string, logger Logger) *CacheStore { + cs := &CacheStore{ + rootPath: rootPath, + ttl: ttl, + maxSize: maxSize, + logger: logger, + entries: make(map[string]*cacheEntry), + routePaths: routePaths, + keyToRoute: make(map[string]string), + keyToPath: make(map[string]string), + accessOrder: []string{}, + maxEntries: 1000, + lastHashes: make(map[string][]byte), + } + + if err := os.MkdirAll(rootPath, 0755); err != nil { + logger.Error("Failed to create cache directory %s: %v", rootPath, err) + } + + cs.loadAll() + + return cs +} + +// key is the cache key for a route+path pair: a hash, so a URL path of any +// shape (including a query string) maps to one safe file name. +func (cs *CacheStore) key(routeName, urlPath string) string { + hash := sha256.Sum256([]byte(routeName + urlPath)) + return fmt.Sprintf("%x", hash[:16]) +} + +// cacheFilePath is the on-disk envelope for a route+path pair. +func (cs *CacheStore) cacheFilePath(routeName, urlPath string) string { + return filepath.Join(cs.rootPath, routeName, cs.key(routeName, urlPath)+".cache") +} + +// load reads one entry from disk, reporting whether it exists and has not +// passed its TTL. +func (cs *CacheStore) load(routeName, urlPath string) (*cacheEntry, bool) { + // urlPath is a URL path: always slash-separated, so clean it with + // path.Clean rather than filepath.Clean, which would rewrite separators + // on Windows and desynchronize keys from the request paths that produced + // them. + urlPath = path.Clean(urlPath) + if urlPath == "." || urlPath == "" { + return nil, false + } + + cacheFile := cs.cacheFilePath(routeName, urlPath) + + data, err := os.ReadFile(cacheFile) + if err != nil { + return nil, false + } + + var entry cacheEntry + if meta, body, ok := parseCacheEnvelope(data); ok && meta.Path == urlPath { + entry = meta + entry.Body = body + } else { + // Legacy two-file layout: body in cacheFile, metadata in cacheFile+".meta" + metaData, err := os.ReadFile(cacheFile + ".meta") + if err != nil { + return nil, false + } + if err := json.Unmarshal(metaData, &entry); err != nil { + return nil, false + } + entry.Body = data + } + + if time.Since(entry.Timestamp) > cs.ttl { + os.Remove(cacheFile) + os.Remove(cacheFile + ".meta") + return nil, false + } + + cs.entriesMu.Lock() + k := cs.key(routeName, urlPath) + if _, exists := cs.entries[k]; !exists { + cs.entries[k] = &entry + cs.keyToRoute[k] = routeName + cs.keyToPath[k] = urlPath + cs.addToAccessOrder(k) + } + cs.entriesMu.Unlock() + + return &entry, true +} + +// Save stores one response for a route+path pair, skipping the write when the +// body is byte-identical to what is already on disk and rejecting bodies over +// the configured size. An unchanged entry still gets its envelope rewritten +// once it is past half its TTL, so refreshing it keeps it alive. +func (cs *CacheStore) Save(routeName, urlPath string, statusCode int, headers http.Header, body []byte) { + urlPath = path.Clean(urlPath) + if urlPath == "." || urlPath == "" { + return + } + + if int64(len(body)) > cs.maxSize { + cs.logger.Error("Response too large for cache (%d bytes) for %s %s", len(body), routeName, urlPath) + return + } + + bodyHash := sha256.Sum256(body) + cacheFile := cs.cacheFilePath(routeName, urlPath) + + cs.entriesMu.Lock() + k := cs.key(routeName, urlPath) + if lastHash, exists := cs.lastHashes[k]; exists && bytes.Equal(lastHash, bodyHash[:]) { + // Unchanged content only allows skipping the write while the cache + // file is still on disk: the file can disappear independently of + // lastHashes (TTL expiry in load, manual cleanup, ...). Skipping the + // write in that case would leave the cache empty forever for + // unchanged responses (e.g. a static model list), sending WOL on + // every offline request. + if info, err := os.Stat(cacheFile); err == nil { + // What expires an entry is the Timestamp inside the envelope, and + // skipping the write leaves it at the original save time. A response + // that never changes would therefore go stale and stop being served + // even though warm-up keeps re-fetching it byte for byte. Refresh the + // envelope once the entry is past half its lifetime instead: that + // bounds the extra writes to roughly one per ttl/2 per entry rather + // than one per warm cycle. + if time.Since(info.ModTime()) < cs.ttl/2 { + cs.entriesMu.Unlock() + cs.logger.Debug("Cache unchanged for %s %s, skipping write", routeName, urlPath) + return + } + } + } + cs.entriesMu.Unlock() + + cacheDir := filepath.Join(cs.rootPath, routeName) + if err := os.MkdirAll(cacheDir, 0755); err != nil { + cs.logger.Error("Failed to create cache directory %s: %v", cacheDir, err) + return + } + + meta := &cacheEntry{ + StatusCode: statusCode, + Headers: sanitizeCachedHeaders(headers), + Body: body, + Timestamp: time.Now(), + Path: urlPath, + } + + // The on-disk entry is a single file: + // + // [4]byte metadata length (little-endian) | metadata JSON | body + // + // Publishing it takes one atomic rename, so a concurrent reader can + // never observe a body without its metadata. With the older two-file + // layout there was a window between the body rename and the metadata + // rename in which a request saw a cache miss and sent WOL for a + // response that was actually being cached. + metaData, err := json.Marshal(meta) + if err != nil { + cs.logger.Error("Failed to marshal cache metadata: %v", err) + return + } + + envelope := make([]byte, 4, 4+len(metaData)+len(body)) + binary.LittleEndian.PutUint32(envelope, uint32(len(metaData))) + envelope = append(envelope, metaData...) + envelope = append(envelope, body...) + + if err := writeFileAtomic(cacheFile, envelope); err != nil { + cs.logger.Error("Failed to write cache file %s: %v", cacheFile, err) + return + } + // Drop a legacy metadata sidecar left behind by older versions. + os.Remove(cacheFile + ".meta") + + cs.entriesMu.Lock() + cs.entries[k] = meta + cs.keyToRoute[k] = routeName + cs.keyToPath[k] = urlPath + cs.addToAccessOrder(k) + cs.lastHashes[k] = bodyHash[:] + cs.entriesMu.Unlock() + + cs.logger.Info("Cache updated for %s %s", routeName, urlPath) +} + +// hopByHopHeaders are connection-specific or framing headers that must not be +// stored and replayed when a cached response is served from disk: the server +// sending the cached response re-adds the correct framing (Content-Length or +// chunking) for the actual body. +var hopByHopHeaders = []string{ + "Connection", + "Keep-Alive", + "Proxy-Authenticate", + "Proxy-Authorization", + "TE", + "Trailer", + "Transfer-Encoding", + "Upgrade", + "Content-Length", +} + +// privateCachedHeaders are response headers that belong to the client that +// fetched them. Replaying one to a different client leaks that client's session, +// so they are dropped on save as well as screened on capture. +var privateCachedHeaders = []string{"Set-Cookie"} + +// sanitizeCachedHeaders strips the framing/hop-by-hop headers that must not +// travel with a stored response, and the per-client headers that must not be +// replayed to anyone else. +func sanitizeCachedHeaders(headers http.Header) http.Header { + sanitized := make(http.Header, len(headers)) + for key, values := range headers { + skip := false + for _, hop := range hopByHopHeaders { + if strings.EqualFold(key, hop) { + skip = true + break + } + } + if !skip { + for _, private := range privateCachedHeaders { + if strings.EqualFold(key, private) { + skip = true + break + } + } + } + if skip { + continue + } + sanitized[key] = values + } + return sanitized +} + +// requestHasUserState reports a request made with credentials, whose response +// is that client's alone: caching it would serve one client's session content +// to every other client that asks for the same path while the machine is down. +func requestHasUserState(r *http.Request) bool { + return r.Header.Get("Authorization") != "" || r.Header.Get("Cookie") != "" +} + +// responseHasUserState reports a response that is private to the client that +// fetched it, either because it sets a cookie or because its Cache-Control +// forbids shared caching. +func responseHasUserState(h http.Header) bool { + if len(h.Values("Set-Cookie")) > 0 { + return true + } + cc := strings.ToLower(h.Get("Cache-Control")) + return strings.Contains(cc, "private") || + strings.Contains(cc, "no-store") || + strings.Contains(cc, "no-cache") +} + +// cacheKeyFor is the cache identity of a request: its path plus its query +// string. /models?filter=x and /models are different responses, so keying on +// the path alone would let whichever was fetched last answer both. A request +// with no query keys on the bare path, which is what the warm-up path stores +// for the configured paths it re-fetches. +func cacheKeyFor(r *http.Request) string { + if r.URL.RawQuery == "" { + return r.URL.Path + } + return r.URL.Path + "?" + r.URL.RawQuery +} + +// splitCacheKey turns a cache key back into the path and query string needed +// to re-fetch it while warming. +func splitCacheKey(key string) (urlPath, rawQuery string) { + if i := strings.Index(key, "?"); i >= 0 { + return key[:i], key[i+1:] + } + return key, "" +} + +// parseCacheEnvelope splits a single-file cache entry +// ([4]byte metadata length (little-endian) | metadata JSON | body). +// ok is false when data is not an envelope (e.g. a legacy two-file body), in +// which case the caller falls back to the legacy layout. +func parseCacheEnvelope(data []byte) (meta cacheEntry, body []byte, ok bool) { + if len(data) < 4 { + return + } + metaLen := int(binary.LittleEndian.Uint32(data[:4])) + if metaLen <= 0 || 4+metaLen > len(data) { + return + } + if err := json.Unmarshal(data[4:4+metaLen], &meta); err != nil { + return + } + if meta.Path == "" { + return + } + meta.Body = data[4+metaLen:] + body = meta.Body + ok = true + return +} + +// writeFileAtomic writes data to a temp file in the same directory and renames +// it into place, so readers never see a partially written file. The temp name is +// unique per write: the wake-triggered warm-up, the periodic warm-up and a +// proxied GET can all save the same key at the same time, and a shared temp path +// would let them interleave into a corrupt envelope before the rename. +func writeFileAtomic(filePath string, data []byte) error { + tmp, err := os.CreateTemp(filepath.Dir(filePath), filepath.Base(filePath)+".*.tmp") + if err != nil { + return err + } + tmpFile := tmp.Name() + + if _, err := tmp.Write(data); err != nil { + tmp.Close() + os.Remove(tmpFile) + return err + } + if err := tmp.Close(); err != nil { + os.Remove(tmpFile) + return err + } + if err := os.Rename(tmpFile, filePath); err != nil { + os.Remove(tmpFile) + return err + } + return nil +} + +// Get returns a stored response as it was saved, or false when there is no +// fresh entry for that route+path pair. +func (cs *CacheStore) Get(routeName, urlPath string) (*http.Response, bool) { + entry, ok := cs.load(routeName, urlPath) + if !ok { + return nil, false + } + + resp := &http.Response{ + StatusCode: entry.StatusCode, + Status: http.StatusText(entry.StatusCode), + Proto: "HTTP/1.1", + ProtoMajor: 1, + ProtoMinor: 1, + Header: entry.Headers, + Body: io.NopCloser(bytes.NewReader(entry.Body)), + } + + return resp, true +} + +// GetCachedPathsForRoute lists the paths worth re-fetching for a route: the +// configured exact paths, plus the cached paths matching a /* pattern. +func (cs *CacheStore) GetCachedPathsForRoute(routeName string) []string { + cs.entriesMu.RLock() + defer cs.entriesMu.RUnlock() + + routePaths, exists := cs.routePaths[routeName] + if !exists || len(routePaths) == 0 { + return nil + } + + var result []string + for _, pattern := range routePaths { + if strings.HasSuffix(pattern, "/*") { + prefix := strings.TrimSuffix(pattern, "/*") + for key := range cs.keyToPath { + if cs.keyToRoute[key] == routeName { + if p := cs.keyToPath[key]; p != "" && strings.HasPrefix(p, prefix) { + result = append(result, p) + } + } + } + } else { + result = append(result, pattern) + } + } + + return result +} + +// addToAccessOrder records a key as most recently used and drops the oldest +// ones once the LRU window overflows. +func (cs *CacheStore) addToAccessOrder(key string) { + for i, k := range cs.accessOrder { + if k == key { + cs.accessOrder = append(cs.accessOrder[:i], cs.accessOrder[i+1:]...) + break + } + } + cs.accessOrder = append(cs.accessOrder, key) + for len(cs.accessOrder) > cs.maxEntries { + oldest := cs.accessOrder[0] + cs.accessOrder = cs.accessOrder[1:] + delete(cs.entries, oldest) + delete(cs.keyToRoute, oldest) + delete(cs.keyToPath, oldest) + } +} + +// loadAll reads the existing cache entries for the configured routes back into +// memory at startup, accepting both the current single-file envelope and the +// older two-file layout. +func (cs *CacheStore) loadAll() { + routes := make(map[string]bool) + for routeName := range cs.routePaths { + routes[routeName] = true + } + + for routeName := range routes { + dir := filepath.Join(cs.rootPath, routeName) + files, err := os.ReadDir(dir) + if err != nil { + continue + } + + for _, file := range files { + if file.IsDir() || !strings.HasSuffix(file.Name(), ".cache") { + continue + } + + if strings.HasSuffix(file.Name(), ".meta.cache") { + continue + } + + cacheFile := filepath.Join(dir, file.Name()) + + var entry cacheEntry + if metaData, err := os.ReadFile(cacheFile + ".meta"); err == nil { + // Legacy two-file layout: body in cacheFile, metadata in the + // ".meta" sidecar + if err := json.Unmarshal(metaData, &entry); err != nil { + continue + } + body, err := os.ReadFile(cacheFile) + if err != nil { + continue + } + entry.Body = body + } else { + // Single-file envelope layout + data, err := os.ReadFile(cacheFile) + if err != nil { + continue + } + if meta, body, ok := parseCacheEnvelope(data); !ok { + continue + } else { + entry = meta + entry.Body = body + } + } + + if time.Since(entry.Timestamp) > cs.ttl { + os.Remove(cacheFile) + os.Remove(cacheFile + ".meta") + continue + } + + k := file.Name()[:len(file.Name())-len(".cache")] + cs.entries[k] = &entry + cs.keyToRoute[k] = routeName + cs.keyToPath[k] = entry.Path + cs.addToAccessOrder(k) + } + } + + cs.logger.Info("Loaded %d cache entries", len(cs.entries)) +} + +// Cleanup drops in-memory entries that have outlived the TTL and removes their +// files, so the cache directory cannot grow without bound. +func (cs *CacheStore) Cleanup() { + cs.entriesMu.Lock() + defer cs.entriesMu.Unlock() + + keysToRemove := []string{} + + for k, entry := range cs.entries { + if time.Since(entry.Timestamp) > cs.ttl { + keysToRemove = append(keysToRemove, k) + } + } + + for _, k := range keysToRemove { + routeName := cs.keyToRoute[k] + urlPath := cs.keyToPath[k] + + delete(cs.entries, k) + delete(cs.keyToRoute, k) + delete(cs.keyToPath, k) + delete(cs.lastHashes, k) + + // Also remove the expired files from disk, not just the in-memory bookkeeping. + if routeName != "" && urlPath != "" { + cacheFile := cs.cacheFilePath(routeName, urlPath) + os.Remove(cacheFile) + os.Remove(cacheFile + ".meta") + } + } + + if len(keysToRemove) > 0 { + cs.logger.Info("Cleaned up %d expired cache entries", len(keysToRemove)) + } +} + +// Close flushes what the cache still holds and clears expired entries. +func (cs *CacheStore) Close() { + cs.Cleanup() +} + +// catchingBody tees the proxied response body into a buffer and hands the +// complete copy to onDone when the body is closed, so the cache is written +// without buffering the whole response in front of the client. +// +// It only hands the copy over when the body ran to EOF without going past limit. +// ReverseProxy closes the upstream body after a copy error too - a client that +// hung up, an upstream reset - and saving that partial body would later serve a +// truncated asset as a full 200. Buffering stops at limit for the same reason +// the check exists: a multi-GB download under a /* pattern must not be held in +// memory on the way to being rejected. +type catchingBody struct { + original io.ReadCloser + buf *bytes.Buffer + limit int64 + overflow bool + eof bool + onDone func(body []byte) + closed bool +} + +// Read streams the body to the client and copies it into the buffer until the +// size limit is passed, at which point the response is no longer worth caching +// and the buffer is dropped. +func (cb *catchingBody) Read(p []byte) (int, error) { + n, err := cb.original.Read(p) + if n > 0 && !cb.overflow { + if int64(cb.buf.Len()+n) > cb.limit { + cb.overflow = true + cb.buf = nil + } else { + cb.buf.Write(p[:n]) + } + } + if err == io.EOF { + cb.eof = true + } + return n, err +} + +// Close hands the buffered body to onDone only if it was complete and within +// the limit, then closes the upstream body. +func (cb *catchingBody) Close() error { + if cb.closed { + return nil + } + cb.closed = true + + if cb.eof && !cb.overflow && cb.onDone != nil { + cb.onDone(cb.buf.Bytes()) + } + if cb.original != nil { + return cb.original.Close() + } + return nil +} + // Health checker: dispatches on the endpoint scheme — tcp:// dials, anything else is an HTTP GET type EndpointHealthChecker struct { client *http.Client @@ -599,8 +1228,12 @@ type ProxyService struct { sshExecutor SSHExecutor logger Logger clock Clock + cache *CacheStore } +// NewProxyService wires the configuration to the health checker, WOL sender, +// SSH executor and (when enabled) the static response cache, and returns the +// service that serves the configured routes. func NewProxyService( config *ProxyConfig, healthChecker HealthChecker, @@ -608,6 +1241,7 @@ func NewProxyService( sshExecutor SSHExecutor, logger Logger, clock Clock, + cache *CacheStore, ) *ProxyService { return &ProxyService{ config: config, @@ -616,6 +1250,7 @@ func NewProxyService( sshExecutor: sshExecutor, logger: logger, clock: clock, + cache: cache, } } @@ -733,10 +1368,167 @@ func (p *ProxyService) checkInactiveMachines() { } } +// warmMachine refreshes the cached paths of every HTTP route pointing at +// machine, fetching each from the route's destination while it is reachable. +func (p *ProxyService) warmMachine(machine *Machine) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + p.logger.Debug("Starting cache warmup for machine: %s", machine.Name) + + client := &http.Client{ + Timeout: 30 * time.Second, + Transport: &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: (&net.Dialer{ + Timeout: 30 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext, + ForceAttemptHTTP2: true, + MaxIdleConns: 10, + IdleConnTimeout: 90 * time.Second, + TLSHandshakeTimeout: 10 * time.Second, + ExpectContinueTimeout: 1 * time.Second, + DisableCompression: true, + }, + } + + for _, route := range routesTo(machine, p.config.Routes) { + if route.IsTCP() { + continue + } + + cachedPaths := p.cache.GetCachedPathsForRoute(route.Name) + if len(cachedPaths) == 0 { + continue + } + + routeURL, err := url.Parse(route.Destination) + if err != nil { + continue + } + + for _, cacheKey := range cachedPaths { + select { + case <-ctx.Done(): + p.logger.Debug("Cache warmup cancelled for machine: %s", machine.Name) + return + case <-time.After(100 * time.Millisecond): + } + + // A stored key carries the query string it was fetched with, so the + // refresh asks for the same URL, not the path alone. + urlPath, rawQuery := splitCacheKey(cacheKey) + reqURL := *routeURL + reqURL.Path = urlPath + reqURL.RawQuery = rawQuery + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil) + if err != nil { + p.logger.Error("Failed to create request for %s %s: %v", machine.Name, cacheKey, err) + continue + } + + resp, err := client.Do(req) + if err != nil { + p.logger.Error("Failed to warm cache for %s %s: %v", machine.Name, cacheKey, err) + continue + } + + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + // One byte past the limit, so an oversized body is *detected* + // rather than silently truncated: LimitReader stops without an + // error, and a cut-off body saved as a 200 would be served to + // every later offline request as if it were the whole asset. + body, err := io.ReadAll(io.LimitReader(resp.Body, p.config.CacheMaxResponseSize+1)) + resp.Body.Close() + + if err != nil { + p.logger.Error("Failed to read %s %s while warming the cache: %v", machine.Name, cacheKey, err) + continue + } + if int64(len(body)) > p.config.CacheMaxResponseSize { + p.logger.Error("Skipping cache for %s %s: response is larger than %d bytes", + machine.Name, cacheKey, p.config.CacheMaxResponseSize) + continue + } + // The warm-up client sends no credentials, but the backend can + // still answer with something that belongs to somebody: a cookie + // or a Cache-Control that says the body is not shareable. Save + // strips Set-Cookie but keeps the body, so the guard has to happen + // here - otherwise a later offline GET is handed a response the + // upstream never meant to give it. + if responseHasUserState(resp.Header) { + p.logger.Debug("Skipping cache for %s %s: response is not shareable", + machine.Name, cacheKey) + continue + } + p.cache.Save(route.Name, cacheKey, resp.StatusCode, resp.Header, body) + } else { + resp.Body.Close() + } + } + } + + p.logger.Debug("Cache warmup completed for machine: %s", machine.Name) +} + +// startCacheCleanup periodically drops expired cache entries. +func (p *ProxyService) startCacheCleanup(ctx context.Context) { + ticker := time.NewTicker(5 * time.Minute) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + p.cache.Cleanup() + } + } +} + +// startCacheWarming re-fetches the cached paths of every route whose machine +// is up, so the cache still holds usable content the next time it sleeps. +func (p *ProxyService) startCacheWarming(ctx context.Context) { + ticker := time.NewTicker(p.config.CacheWarmInterval) + defer ticker.Stop() + + for _, machine := range p.config.Machines { + machine.mu.RLock() + isHealthy := machine.IsHealthy + machine.mu.RUnlock() + if isHealthy { + p.warmMachine(machine) + } + } + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + for _, machine := range p.config.Machines { + machine.mu.RLock() + isHealthy := machine.IsHealthy + machine.mu.RUnlock() + + if isHealthy { + p.warmMachine(machine) + } + } + } + } +} + +// isSecureServer reports a config that should serve TLS rather than plain HTTP. func isSecureServer(config *ProxyConfig) bool { return config.SSLCertificate != "" && config.SSLCertificateKey != "" } +// Start brings the service up: background health checks, the HTTP/TCP route +// listeners, cache maintenance and the idle-shutdown sweep. It returns when ctx +// is cancelled and the listeners have drained. func (p *ProxyService) Start(ctx context.Context) error { // Start background health checks p.healthChecker.StartBackgroundChecks( @@ -755,6 +1547,13 @@ func (p *ProxyService) Start(ctx context.Context) error { // Start background inactivity monitor go p.startInactivityMonitor(ctx) + if p.cache != nil { + p.logger.Info("Cache enabled, root path: %s, TTL: %v, warm interval: %v", + p.config.CacheRootPath, p.config.CacheTTL, p.config.CacheWarmInterval) + go p.startCacheCleanup(ctx) + go p.startCacheWarming(ctx) + } + p.logger.Info("Initial health checks completed, starting HTTP server") // Log configured routes @@ -824,6 +1623,10 @@ func (p *ProxyService) Start(ctx context.Context) error { cancelServers() } } + + if p.cache != nil { + p.cache.Close() + } return firstErr } @@ -891,6 +1694,9 @@ func clientAddr(r *http.Request) string { return r.RemoteAddr } +// handleRequest serves one request on one route: resolve the route, log where +// the request came from, answer from the static cache if the machine is +// confirmed down and has a usable entry, and otherwise wake the machine first. func (p *ProxyService) handleRequest(w http.ResponseWriter, r *http.Request) { route := p.routeForRequest(r) @@ -914,6 +1720,35 @@ func (p *ProxyService) handleRequest(w http.ResponseWriter, r *http.Request) { return } + // The machine looks down: before waking it, see whether the static cache + // can answer. A cached response means the client gets a full answer + // while the machine stays asleep. + // + // Only a machine confirmed unhealthy may be answered from cache. The + // health cache going stale is not evidence the machine is down, and a live + // machine must serve its own response rather than a stored one. + // + // A request that carries credentials is not answered from cache either: + // every stored entry is one that was fetched anonymously, so replaying it + // to a logged-in client would give that client a response the backend + // never produced for it. + if p.cache != nil && r.Method == http.MethodGet && + machineConfirmedDown(machineState) && !requestHasUserState(r) { + if resp, ok := p.cache.Get(route.Name, cacheKeyFor(r)); ok { + p.logger.Info("Serving cached response for %s %s (machine %s is down)", + r.Method, r.URL.RequestURI(), machineName) + for k, v := range resp.Header { + for _, header := range v { + w.Header().Add(k, header) + } + } + w.WriteHeader(resp.StatusCode) + io.Copy(w, resp.Body) + resp.Body.Close() + return + } + } + // Need to wake up the server p.logger.Info("Machine %s appears down (%s), waking for %s", machineName, reason, route.Name) p.healthChecker.CloseIdleConnections() @@ -1016,6 +1851,18 @@ func (p *ProxyService) routeForRequest(r *http.Request) *Route { return p.config.RoutesByHostname[host] } +// machineConfirmedDown reports a machine the health checker has positively seen +// unhealthy. It is deliberately narrower than "the health cache says no": a +// healthy machine whose entry has merely gone stale is not down. +func machineConfirmedDown(machine *Machine) bool { + machine.mu.RLock() + defer machine.mu.RUnlock() + return !machine.IsHealthy +} + +// healthCacheStatus reports whether the machine's health entry is fresh enough +// to act on, and why not when it is not. Note that "not cached" covers both an +// unhealthy machine and a healthy machine whose entry has simply gone stale. func (p *ProxyService) healthCacheStatus(machine *Machine) (cached bool, reason string) { machine.mu.RLock() defer machine.mu.RUnlock() @@ -1079,6 +1926,8 @@ func (p *ProxyService) wakeAndWait(ctx context.Context, machine *Machine) error return p.waitForWake(ctx, machine) } +// waitForWake sends magic packets on a ticker and polls health until the machine +// answers or the wake timeout expires. func (p *ProxyService) waitForWake(ctx context.Context, machine *Machine) error { timeout := p.clock.After(p.config.Timeout) healthCheckTicker := p.clock.NewTicker(p.config.PollInterval) @@ -1122,6 +1971,19 @@ func (p *ProxyService) waitForWake(ctx context.Context, machine *Machine) error wakeDuration := p.clock.Now().Sub(wakeStartTime) p.logger.Info("Machine %s woke up after %v", machine.Name, wakeDuration) + + if p.cache != nil { + // Refresh the cached paths now that the machine is + // reachable again. Only invalidating the cache and + // relying on the next warm tick (up to warm_interval + // later) or a future proxied GET to repopulate it means + // a machine that sleeps again before either happens + // falls back to triggering WOL on every cached request. + // Refreshing in place (instead of deleting first) also + // keeps the previous entries servable if the refresh + // itself fails. + go p.warmMachine(machine) + } return nil } } @@ -1244,6 +2106,8 @@ func (p *ProxyService) forwardTCPConn(ctx context.Context, client net.Conn, rout } } +// proxyRequest forwards one request upstream and records the activity on the +// machine, so an idle timeout is not counted while a client is being served. func (p *ProxyService) proxyRequest(w http.ResponseWriter, r *http.Request, route *Route) { machineState := route.Machine machineState.mu.Lock() @@ -1314,6 +2178,41 @@ func (p *ProxyService) proxyRequest(w http.ResponseWriter, r *http.Request, rout proxy.ModifyResponse = func(resp *http.Response) error { p.logger.Info("Response from %s: status=%d, content-length=%d", route.Name, resp.StatusCode, resp.ContentLength) + + // Only a 2xx GET on a configured path, from an anonymous request, to a + // response that is not marked private: anything else belongs to the one + // client that fetched it and must not be replayed to the next caller. + if p.cache != nil && r.Method == http.MethodGet && resp.StatusCode >= 200 && resp.StatusCode < 300 && + !requestHasUserState(r) && !responseHasUserState(resp.Header) { + cachePaths, exists := p.config.CacheRoutes[route.Name] + if exists { + var shouldCache bool + for _, pattern := range cachePaths { + if strings.HasSuffix(pattern, "/*") { + prefix := strings.TrimSuffix(pattern, "/*") + if strings.HasPrefix(r.URL.Path, prefix) { + shouldCache = true + break + } + } else if r.URL.Path == pattern { + shouldCache = true + break + } + } + + if shouldCache { + resp.Body = &catchingBody{ + original: resp.Body, + buf: &bytes.Buffer{}, + limit: p.config.CacheMaxResponseSize, + onDone: func(body []byte) { + p.cache.Save(route.Name, cacheKeyFor(r), resp.StatusCode, resp.Header, body) + }, + } + } + } + } + return nil } @@ -1411,6 +2310,58 @@ func LoadConfig(filename string, clock Clock) (*ProxyConfig, error) { return nil, fmt.Errorf("invalid health_cache_duration: %w", err) } + var cacheEnabled bool + var cacheRootPath string + var cacheTTL time.Duration + var cacheMaxResponseSize int64 + var cacheWarmInterval time.Duration + cacheRoutes := make(map[string][]string) + + if config.CacheConfig.Enabled { + cacheEnabled = true + cacheRootPath = config.CacheConfig.RootPath + if cacheRootPath == "" { + cacheRootPath = "./cache" + } + + cacheTTL = 24 * time.Hour + if config.CacheConfig.TTL != "" { + cacheTTL, err = time.ParseDuration(config.CacheConfig.TTL) + if err != nil { + return nil, fmt.Errorf("invalid cache ttl: %w", err) + } + // A non-positive ttl expires every entry the moment it is written, so + // the cache never answers a single request while looking configured. + if cacheTTL <= 0 { + return nil, fmt.Errorf("cache ttl must be positive, got %v", cacheTTL) + } + } + + cacheMaxResponseSize = 10 * 1024 * 1024 + if config.CacheConfig.MaxResponseSizeBytes > 0 { + cacheMaxResponseSize = config.CacheConfig.MaxResponseSizeBytes + } + + cacheWarmInterval = 30 * time.Minute + if config.CacheConfig.WarmInterval != "" { + cacheWarmInterval, err = time.ParseDuration(config.CacheConfig.WarmInterval) + if err != nil { + return nil, fmt.Errorf("invalid cache warm_interval: %w", err) + } + // time.NewTicker panics on a non-positive interval, and it panics + // inside the warming goroutine, which takes the whole proxy down at + // startup. An operator writing "0s" to mean "don't warm" must get a + // configuration error instead. + if cacheWarmInterval <= 0 { + return nil, fmt.Errorf("cache warm_interval must be positive, got %v", cacheWarmInterval) + } + } + + for _, cr := range config.CacheConfig.Routes { + cacheRoutes[cr.Route] = cr.Paths + } + } + if len(config.Targets) > 0 && (len(config.Machines) > 0 || len(config.Routes) > 0) { return nil, fmt.Errorf("config mixes the legacy [[targets]] format with [[machines]]/[[routes]]; use one or the other") } @@ -1426,6 +2377,25 @@ func LoadConfig(filename string, clock Clock) (*ProxyConfig, error) { return nil, err } + // A cache block naming a route that does not exist is a typo that would + // otherwise silently cache nothing. TCP routes carry no HTTP semantics, + // so they cannot be cached either. + for _, cr := range config.CacheConfig.Routes { + found := false + for _, route := range model.routes { + if route.Name == cr.Route { + if route.IsTCP() { + return nil, fmt.Errorf("cache route %q is a TCP route; only HTTP routes can be cached", cr.Route) + } + found = true + break + } + } + if !found { + return nil, fmt.Errorf("cache references unknown route %q", cr.Route) + } + } + return &ProxyConfig{ Port: config.Port, SSLCertificate: config.SSLCertificate, @@ -1443,6 +2413,12 @@ func LoadConfig(filename string, clock Clock) (*ProxyConfig, error) { Routes: model.routes, RoutesByHostname: model.byHostname, RoutesByListenPort: model.byListenPort, + CacheEnabled: cacheEnabled, + CacheRootPath: cacheRootPath, + CacheTTL: cacheTTL, + CacheMaxResponseSize: cacheMaxResponseSize, + CacheWarmInterval: cacheWarmInterval, + CacheRoutes: cacheRoutes, }, nil } @@ -1776,6 +2752,13 @@ func (l *StdLogger) Error(msg string, args ...interface{}) { log.Printf("[ERROR] "+msg, args...) } +// Debug writes a debug-level line. +func (l *StdLogger) Debug(msg string, args ...interface{}) { + log.Printf("[DEBUG] "+msg, args...) +} + +// logStartup announces the running build, so a log identifies the version that +// produced it. func logStartup(logger Logger, version string) { logger.Info("Starting doormouse version `%s`", version) } @@ -1838,8 +2821,13 @@ func main() { wolSender := NewUDPWOLSender(logger) sshExecutor := NewDefaultSSHExecutor(logger) + var cache *CacheStore + if config.CacheEnabled { + cache = NewCacheStore(config.CacheRootPath, config.CacheTTL, config.CacheMaxResponseSize, config.CacheRoutes, logger) + } + // Create proxy service - proxy := NewProxyService(config, healthChecker, wolSender, sshExecutor, logger, clock) + proxy := NewProxyService(config, healthChecker, wolSender, sshExecutor, logger, clock, cache) // Start the service, shutting down cleanly on SIGINT/SIGTERM ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) diff --git a/main_test.go b/main_test.go index 75445f4..9506624 100644 --- a/main_test.go +++ b/main_test.go @@ -313,6 +313,10 @@ func (l *recordingLogger) Info(msg string, args ...interface{}) { func (l *recordingLogger) Error(msg string, args ...interface{}) { l.Info(msg, args...) } +// Debug records the debug line alongside the rest, so a test can assert on it. +func (l *recordingLogger) Debug(msg string, args ...interface{}) { l.Info(msg, args...) } + +// all returns everything the logger has seen, newest last. func (l *recordingLogger) all() string { l.mu.Lock() defer l.mu.Unlock() @@ -321,9 +325,15 @@ func (l *recordingLogger) all() string { type noopLogger struct{} -func (noopLogger) Info(_ string, _ ...interface{}) {} +// Info discards the message: most tests do not care about logging. +func (noopLogger) Info(_ string, _ ...interface{}) {} + +// Error discards the message. func (noopLogger) Error(_ string, _ ...interface{}) {} +// Debug discards the message. +func (noopLogger) Debug(_ string, _ ...interface{}) {} + type startupLogger struct { info []string error []string @@ -337,6 +347,11 @@ func (l *startupLogger) Error(msg string, args ...interface{}) { l.error = append(l.error, fmt.Sprintf(msg, args...)) } +// Debug ignores debug lines: this logger only captures what startup logged. +func (l *startupLogger) Debug(msg string, args ...interface{}) {} + +// TestLogStartup_LogsDoormouseVersionAtInfoLevel checks that the running build +// is announced at info level on startup. func TestLogStartup_LogsDoormouseVersionAtInfoLevel(t *testing.T) { logger := &startupLogger{} @@ -425,7 +440,7 @@ func newTestProxy(t *testing.T, backendHandler http.Handler) *testProxy { RoutesByHostname: map[string]*Route{testHostname: route}, } - svc := NewProxyService(cfg, health, wol, ssh, noopLogger{}, clock) + svc := NewProxyService(cfg, health, wol, ssh, noopLogger{}, clock, nil) return &testProxy{svc: svc, clock: clock, health: health, wol: wol, ssh: ssh, machine: machineState, route: route, backend: backend} }