diff --git a/dash/api.go b/dash/api.go index 60535a9a..6a6a1212 100644 --- a/dash/api.go +++ b/dash/api.go @@ -107,13 +107,16 @@ const dashCacheTTL = 30 * time.Second // dashCacheStale is how far PAST the TTL a body may still be served while its replacement is being // computed. Beyond it a reader waits for fresh numbers instead. // -// The cap is what keeps stale-while-revalidate honest. Without one, a deployment that goes quiet -// serves the last body it ever built, indefinitely, and the page reports it as current — the -// dashboard's freshness line is stamped when the BROWSER fetched, so it cannot see server-side -// staleness. 30s past a 30s TTL bounds the worst case at one minute, which is smaller than the -// client's own refresh interval, so no reader can be shown anything older than they would have -// been shown anyway by not clicking refresh. -const dashCacheStale = 30 * time.Second +// This is the cap that keeps stale-while-revalidate honest, and it was 30s, which was SHORTER THAN +// THE WORK AGAIN: an all-time stats read takes about 6s warm and 35s on a cold process on a 3.6 GB +// database, so a reader who came back after a minute found nothing to serve and paid the whole +// query, every time. Ten minutes covers the gap between two looks at a dashboard. +// +// The honesty that the old comment asked of the cap now comes from the response instead of from a +// short cap: every cached body carries an Age header (seconds since it was computed), the page +// subtracts it from "Updated N ago", and the reader sees the age of the data rather than the age of +// the request. Nothing is served older than this, and nothing is shown as newer than it is. +const dashCacheStale = 10 * time.Minute // dashHandlerTimeout bounds every dashboard read except the four in unboundedRoutes. It used to // bound three of them, named by hand at their route entries — see Mount, which now applies it by @@ -176,6 +179,8 @@ type jsonCache struct { // serveJSON — which is both wasted work and the thing that was turning an isolated ~1s query // into a 40s one under real concurrent load. refreshing map[string]bool + // gen counts clear() calls; see generation. + gen uint64 } type jsonCacheEntry struct { @@ -183,6 +188,44 @@ type jsonCacheEntry struct { body []byte } +// clear drops every entry. A refresh already in flight is allowed to finish; it stores a body computed +// after the clear began only if it started after, so the claim is released but the old entry is gone. +func (c *jsonCache) clear() { + c.mu.Lock() + defer c.mu.Unlock() + c.entries = nil + c.gen++ +} + +// generation is read before a compute starts and handed to setIfCurrent after it: a compute that began +// before a clear read state that has since changed, and its body must not be stored as current. +func (c *jsonCache) generation() uint64 { + c.mu.Lock() + defer c.mu.Unlock() + return c.gen +} + +// setIfCurrent stores body only if no clear() has happened since gen was read. The check and the store are +// ONE lock hold: with two, a clear landing between them put the old body back. +func (c *jsonCache) setIfCurrent(key string, body []byte, gen uint64) { + c.mu.Lock() + defer c.mu.Unlock() + if c.gen == gen { + c.setLocked(key, body) + } +} + +// age is how long ago the entry for key was computed; 0 when there is none. +func (c *jsonCache) age(key string) time.Duration { + c.mu.Lock() + defer c.mu.Unlock() + e, ok := c.entries[key] + if !ok { + return 0 + } + return time.Since(e.at) +} + func (c *jsonCache) get(key string) ([]byte, bool) { c.mu.Lock() defer c.mu.Unlock() @@ -204,17 +247,24 @@ func (c *jsonCache) get(key string) ([]byte, bool) { // endRefresh runs (via defer) the moment it returns either way, so this cannot serve a body // forever — only up to dashCacheTTL+dashCacheStale+dashComputeTimeout past when it was written. func (c *jsonCache) load(key string) (body []byte, fresh bool) { + body, fresh, _ = c.loadAged(key) + return body, fresh +} + +// loadAged is load plus the entry's age, read under the same lock hold as the body so the pair is +// consistent: a clear between two separate reads made an old body report an age of 0. +func (c *jsonCache) loadAged(key string) (body []byte, fresh bool, age time.Duration) { c.mu.Lock() defer c.mu.Unlock() e, ok := c.entries[key] if !ok { - return nil, false + return nil, false, 0 } - age := time.Since(e.at) + age = time.Since(e.at) if age >= dashCacheTTL+dashCacheStale && !c.refreshing[key] { - return nil, false + return nil, false, 0 } - return e.body, age < dashCacheTTL + return e.body, age < dashCacheTTL, age } // beginRefresh claims the right to refresh key, returning false if another refresh for it is @@ -242,6 +292,10 @@ func (c *jsonCache) endRefresh(key string) { func (c *jsonCache) set(key string, body []byte) { c.mu.Lock() defer c.mu.Unlock() + c.setLocked(key, body) +} + +func (c *jsonCache) setLocked(key string, body []byte) { if c.entries == nil { c.entries = make(map[string]jsonCacheEntry) } @@ -266,17 +320,20 @@ func (c *jsonCache) set(key string, body []byte) { // that no code change produced. The header costs one line per response and makes the harness // able to tell the two apart. func serveCached(w http.ResponseWriter, c *jsonCache, key string) bool { - if body, ok := c.get(key); ok { - writeCachedJSON(w, body, "hit") + if body, fresh, age := c.loadAged(key); body != nil && fresh { + writeCachedJSON(w, body, "hit", age) return true } w.Header().Set("X-Cache", "miss") return false } -func writeCachedJSON(w http.ResponseWriter, body []byte, state string) { +func writeCachedJSON(w http.ResponseWriter, body []byte, state string, age time.Duration) { w.Header().Set("Content-Type", "application/json") w.Header().Set("X-Cache", state) + // Age (RFC 9111) is how old the body is, so the UI can say how old the figures are. 0 for a body + // computed for this request. + w.Header().Set("Age", strconv.FormatInt(int64(age/time.Second), 10)) _, _ = w.Write(body) } @@ -304,9 +361,9 @@ func writeCachedJSON(w http.ResponseWriter, body []byte, state string) { // be killed instantly and the entry could never become fresh again. func (a *API) serveJSON(w http.ResponseWriter, r *http.Request, c *jsonCache, key string, compute func(db *DB) ([]byte, error)) { - if body, fresh := c.load(key); body != nil { + if body, fresh, age := c.loadAged(key); body != nil { if fresh { - writeCachedJSON(w, body, "hit") + writeCachedJSON(w, body, "hit", age) return } // Only the first stale reader for this key starts a refresh; every other concurrent @@ -319,10 +376,11 @@ func (a *API) serveJSON(w http.ResponseWriter, r *http.Request, c *jsonCache, ke if c.beginRefresh(key) { go a.refreshJSON(c, key, compute) } - writeCachedJSON(w, body, "stale") + writeCachedJSON(w, body, "stale", age) return } w.Header().Set("X-Cache", "miss") + w.Header().Set("Age", "0") // The shared computation is DETACHED from every caller, and that is deliberate rather than // careless. Bound to the leader's r.Context() it inherits the leader's deadline and the leader's // abandonment: when the leader timed out, compute failed with a cancelled context and @@ -340,9 +398,10 @@ func (a *API) serveJSON(w http.ResponseWriter, r *http.Request, c *jsonCache, ke v, err, _ := a.jsonInflight.Do(fmt.Sprintf("%p\x00%s", c, key), func() (any, error) { ctx, cancel := context.WithTimeout(context.Background(), dashComputeTimeout) defer cancel() + gen := c.generation() body, err := compute(a.rec.DB().WithContext(ctx)) if err == nil { - c.set(key, body) + c.setIfCurrent(key, body, gen) } return body, err }) @@ -390,6 +449,20 @@ func (c *capture) WriteHeader(code int) { c.code = code } // on each tab switch and on its five-minute refresh; none of them existed in a cache before, so // a reader paid the full price every time. The key is the caller's principal plus the full query // string (cacheKey), so two accounts never share a body, and the path, so two routes never do. +// InvalidateStateCaches drops cached reads that include state a user can edit, so a change shows on +// the next load instead of after the stale window. kind is the tenant package's ChangeAccount (an +// account's configuration, which holds the tool and skill removal list) or ChangeStrategy (keep-alive +// strategies and campaigns). Read-only aggregates (stats, series, sessions) are left to the stale-while- +// revalidate path: nothing a user can edit appears in them. +func (a *API) InvalidateStateCaches(kind string) { + a.toolsCache.clear() + a.toolFilterCache.clear() + a.promptCache.clear() + if kind == "strategy" { + a.routeCache.clear() + } +} + func (a *API) cachedRoute(h http.HandlerFunc) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { p := Principal{Manager: true} @@ -425,6 +498,7 @@ func (a *API) refreshJSON(c *jsonCache, key string, compute func(db *DB) ([]byte defer c.endRefresh(key) ctx, cancel := context.WithTimeout(context.Background(), dashComputeTimeout) defer cancel() + gen := c.generation() body, err := compute(a.rec.DB().WithContext(ctx)) if err != nil { // Keep the stale entry: it is still servable and still inside its cap. Dropping it would @@ -433,7 +507,7 @@ func (a *API) refreshJSON(c *jsonCache, key string, compute func(db *DB) ([]byte "err", err) return } - c.set(key, body) + c.setIfCurrent(key, body, gen) } // cacheKey scopes a cache entry to the actual caller, not just the URL: a manager and a diff --git a/dash/jsoncache_age_test.go b/dash/jsoncache_age_test.go new file mode 100644 index 00000000..e020b13a --- /dev/null +++ b/dash/jsoncache_age_test.go @@ -0,0 +1,173 @@ +package dash + +import ( + "net/http" + "net/http/httptest" + "os" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" +) + +// A body older than the 30 s freshness window but inside the stale cap is served at once, labelled +// stale, with its real age, and exactly one refresh runs behind it. Before the cap was widened a +// reader who came back after a minute found nothing to serve and paid the whole query. +func TestServeJSONServesAStaleBodyWithItsAge(t *testing.T) { + a, _ := newTestAPI(t, Options{}) + c := &jsonCache{entries: map[string]jsonCacheEntry{ + "k": {at: time.Now().Add(-5 * time.Minute), body: []byte(`{"gen":0}`)}, + }} + var computes int32 + refreshed := make(chan struct{}) + compute := func(*DB) ([]byte, error) { + atomic.AddInt32(&computes, 1) + close(refreshed) + return []byte(`{"gen":1}`), nil + } + w := httptest.NewRecorder() + a.serveJSON(w, httptest.NewRequest(http.MethodGet, "/api/stats", nil), c, "k", compute) + + if got := w.Header().Get("X-Cache"); got != "stale" { + t.Fatalf("X-Cache = %q, want stale: a 5-minute-old body is inside the cap and must be served, not recomputed", got) + } + if w.Body.String() != `{"gen":0}` { + t.Errorf("body = %q, want the cached one", w.Body.String()) + } + age, err := strconv.Atoi(w.Header().Get("Age")) + if err != nil || age < 299 || age > 305 { + t.Errorf("Age = %q, want about 300 seconds", w.Header().Get("Age")) + } + select { + case <-refreshed: + case <-time.After(3 * time.Second): + t.Fatal("no refresh was started behind the stale answer") + } + deadline := time.Now().Add(2 * time.Second) + for { + if body, fresh := c.load("k"); fresh && string(body) == `{"gen":1}` { + break + } + if time.Now().After(deadline) { + t.Fatal("the refreshed body never replaced the stale one") + } + time.Sleep(time.Millisecond) + } + if n := atomic.LoadInt32(&computes); n != 1 { + t.Errorf("compute ran %d times, want 1", n) + } +} + +// Past the cap nothing is served: the reader waits for current figures, and the answer says Age 0. +func TestServeJSONComputesWhenPastTheCap(t *testing.T) { + a, _ := newTestAPI(t, Options{}) + c := &jsonCache{entries: map[string]jsonCacheEntry{ + "k": {at: time.Now().Add(-(dashCacheTTL + dashCacheStale + time.Minute)), body: []byte(`{"gen":0}`)}, + }} + w := httptest.NewRecorder() + a.serveJSON(w, httptest.NewRequest(http.MethodGet, "/api/stats", nil), c, "k", + func(*DB) ([]byte, error) { return []byte(`{"gen":2}`), nil }) + if w.Body.String() != `{"gen":2}` || w.Header().Get("X-Cache") != "miss" || w.Header().Get("Age") != "0" { + t.Errorf("got %q X-Cache=%q Age=%q, want a fresh compute labelled miss with Age 0", + w.Body.String(), w.Header().Get("X-Cache"), w.Header().Get("Age")) + } +} + +// The page must show the age of the data. These are source checks like the neighbouring UI tests: +// the Age header is read into state.ageMs and Overview subtracts it from "Updated N ago". +func TestOverviewFreshnessUsesTheAgeOfTheData(t *testing.T) { + js, err := os.ReadFile("ui/app.js") + if err != nil { + t.Fatal(err) + } + src := string(js) + for _, want := range []string{ + "res.headers.get('Age')", + "state.loadedAt = Date.now() - dataAge", + } { + if !strings.Contains(src, want) { + t.Errorf("app.js no longer contains %q; \"Updated N ago\" would claim fresher data than it shows", want) + } + } +} + +// InvalidateStateCaches drops the cached tool reads at once, so a write shows on the next load; a read +// that began before the write and finishes after it must not put its old body back. +func TestInvalidateStateCachesDropsReadsAndIgnoresALateOldCompute(t *testing.T) { + a, _ := newTestAPI(t, Options{}) + a.toolFilterCache.set("k", []byte(`{"v":"before the click"}`)) + a.routeCache.set("r", []byte(`{"v":"route"}`)) + + gen := a.toolFilterCache.generation() // a compute begins now, reading the old state + a.InvalidateStateCaches("account") + a.toolFilterCache.setIfCurrent("k", []byte(`{"v":"computed from the old state"}`), gen) + + if body, _ := a.toolFilterCache.load("k"); body != nil { + t.Errorf("after the write the cache still holds %q", body) + } + if body, _ := a.routeCache.load("r"); body == nil { + t.Error("an account write dropped the route cache, which holds no account state") + } + a.InvalidateStateCaches("strategy") + if body, _ := a.routeCache.load("r"); body != nil { + t.Errorf("a strategy write left the route cache holding %q", body) + } +} + +// The live routes go through serveJSON, so this is the guard on the real path: a clear that lands while +// a compute is running must not be undone by that compute finishing, in the foreground flight and in the +// background refresh alike. The next read after the write recomputes from the new state. +func TestAClearDuringASlowComputeIsNotUndoneByIt(t *testing.T) { + a, _ := newTestAPI(t, Options{}) + c := &a.toolFilterCache + + // foreground flight: a cold read starts, the write lands, the read finishes. + release := make(chan struct{}) + started := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + a.serveJSON(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/api/toolfilter", nil), c, "k", + func(*DB) ([]byte, error) { close(started); <-release; return []byte(`{"v":"old state"}`), nil }) + }() + <-started + a.InvalidateStateCaches("account") + close(release) + <-done + if body, _ := c.load("k"); body != nil { + t.Fatalf("a compute that began before the write stored %q after it", body) + } + + // background refresh: a stale entry, the refresh starts, the write lands, the refresh finishes. + c.set("k2", []byte(`{"v":"before"}`)) + c.mu.Lock() + e := c.entries["k2"] + e.at = time.Now().Add(-2 * time.Minute) + c.entries["k2"] = e + c.mu.Unlock() + release2 := make(chan struct{}) + started2 := make(chan struct{}) + w := httptest.NewRecorder() + a.serveJSON(w, httptest.NewRequest(http.MethodGet, "/api/toolfilter", nil), c, "k2", + func(*DB) ([]byte, error) { close(started2); <-release2; return []byte(`{"v":"old state"}`), nil }) + <-started2 + a.InvalidateStateCaches("account") + close(release2) + deadline := time.Now().Add(2 * time.Second) + for { + c.mu.Lock() + refreshing := c.refreshing["k2"] + c.mu.Unlock() + if !refreshing { + break + } + if time.Now().After(deadline) { + t.Fatal("the refresh never finished") + } + time.Sleep(time.Millisecond) + } + if body, _ := c.load("k2"); body != nil { + t.Fatalf("a refresh that began before the write stored %q after it", body) + } +} diff --git a/dash/ui/app.js b/dash/ui/app.js index 61e8bca0..49e25336 100644 --- a/dash/ui/app.js +++ b/dash/ui/app.js @@ -319,6 +319,10 @@ async function api(path, extra) { try { const j = await res.json(); if (j.error) msg = j.error; } catch (_) { /* not json */ } const e = new Error(msg); e.status = res.status; throw e; } + // The server may answer from a cache that is up to ten minutes old (stale-while-revalidate) and + // says how old in the Age header. Keep it per endpoint so a page can show the age of its DATA. + const age = Number(res.headers.get('Age')) || 0; + (state.ageMs ||= {})[path] = age * 1000; return res.json(); } @@ -1559,7 +1563,9 @@ async function loadOverview(opts = {}) { const lean = await api('stats', { lean: 1 }); state.overview = lean; state.overviewFull = false; - state.loadedAt = Date.now(); + // "Updated N ago" is the age of the DATA: a body served from the server's stale cache is older + // than the moment it arrived (see the Age header read in api()). + state.loadedAt = Date.now() - ((state.ageMs && state.ageMs.stats) || 0); if (state.dirty === wasDirty) { state.dirty = false; $('#refresh-now').classList.remove('has-new'); @@ -1574,6 +1580,13 @@ async function loadOverview(opts = {}) { const o = await api('stats'); state.overview = o; state.overviewFull = true; + // Past the 30 s freshness window the server is recomputing behind a stale answer, so look again once, + // shortly, and the page converges on current figures unasked. + const dataAge = (state.ageMs && state.ageMs.stats) || 0; + state.loadedAt = Date.now() - dataAge; + if (dataAge > 45000 && !state.staleRetry) { + state.staleRetry = setTimeout(() => { state.staleRetry = 0; if (state.view === 'overview') loadOverview(); }, 10000); + } renderSavedLedger(o, state.keepAliveLedger, false); if ($('#overview-diagnostics').open) renderDiagnostics(); else state.diagStale = true; diff --git a/proxy/adaptapply.go b/proxy/adaptapply.go index 7e4f005a..8c190666 100644 --- a/proxy/adaptapply.go +++ b/proxy/adaptapply.go @@ -294,6 +294,11 @@ func (h *Handler) reloadLocal(doc string) error { t := *cur t.Pipe, t.Preset = built.Pipe, built.Preset h.static.Store(&t) + // A single-tenant proxy has no registry to announce the change, so say it here: the removal list in + // the running pipeline just changed, and a cached dashboard read of it is now wrong. + if h.api != nil { + h.api.InvalidateStateCaches(tenant.ChangeAccount) + } return nil } diff --git a/proxy/adaptctl_test.go b/proxy/adaptctl_test.go index a975e083..fce808cc 100644 --- a/proxy/adaptctl_test.go +++ b/proxy/adaptctl_test.go @@ -428,3 +428,33 @@ func TestSaveMoreAppliesThePresetWithoutOverrides(t *testing.T) { t.Fatalf("stored %q, priced %q", doc, want) } } + +// On a single-tenant proxy there is no registry to announce a configuration change, so a local apply or +// revert has to drop the dashboard's cached reads itself: the removal list in the running pipeline just +// changed, and /api/toolfilter may otherwise answer from a body up to ten minutes old. +func TestAdaptApplyLocalDropsTheDashboardsCachedReads(t *testing.T) { + h := localAdaptHandler(t, filepath.Join(t.TempDir(), "d.db.adapt.json")) + mux := h.Mux() + get := func() string { + r := httptest.NewRequest("GET", "/api/toolfilter", nil) + r.RemoteAddr = "127.0.0.1:4000" + w := httptest.NewRecorder() + mux.ServeHTTP(w, r) + return w.Header().Get("X-Cache") + } + get() // computes and caches + if got := get(); got != "hit" { + t.Fatalf("second read X-Cache = %q, want hit; the test needs a warm cache to prove anything", got) + } + basis := offer(h, "", "preset: house\n", []string{"format", "searchfold", "textclean"}, nil) + r := httptest.NewRequest("POST", "/api/adapt/compaction/apply", strings.NewReader(`{"option":"c1","basis":"`+basis+`"}`)) + r.RemoteAddr = "127.0.0.1:4000" + w := httptest.NewRecorder() + mux.ServeHTTP(w, r) + if w.Code != http.StatusOK { + t.Fatalf("apply = %d %s", w.Code, w.Body) + } + if got := get(); got == "hit" || got == "stale" { + t.Errorf("read right after a local apply X-Cache = %q; a cached body from before the apply was served", got) + } +} diff --git a/proxy/proxy.go b/proxy/proxy.go index ef1180b0..a252237f 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -362,6 +362,11 @@ func New(pipe *components.Pipeline, st store.Store, agg *metrics.Aggregator, opt } if h.rec != nil { h.api = dash.NewAPI(h.rec) + // A write to an account or to the keep-alive strategies drops the cached dashboard reads that + // include that state, so a toggle shows on the next load and not after the stale window. + if reg := h.registry(); reg != nil { + reg.SetOnChange(h.api.InvalidateStateCaches) + } // Publish the off-path pool's counters to the dashboard, the same layering /stats // uses: the pool lives in `modes`, which sits above both `metrics` and `dash`, so // the host is the only place that can join them. Read at serve time, not captured diff --git a/proxy/toolfilterctl_test.go b/proxy/toolfilterctl_test.go index 7e851c88..9d0441c5 100644 --- a/proxy/toolfilterctl_test.go +++ b/proxy/toolfilterctl_test.go @@ -393,3 +393,39 @@ func TestToolFilterStoresASkillUnderItsOwnPrefix(t *testing.T) { t.Errorf("re-including left a skill in the removal list: %q", doc) } } + +// A removal switch has to show on the very next load. GET /api/toolfilter is served from the +// dashboard's response cache, which may answer from a body up to ten minutes old; the write must +// therefore drop that cached body, or a reload would show the switch as it was before the click. +func TestToolFilterWriteShowsOnTheNextRead(t *testing.T) { + f := newMgrFixture(t) + f.h.API().SetToolFilterState(f.h.DashToolFilter()) + w, _ := f.signUp(t, "boss@ibm.com", "l") + jar := w.Result().Cookies() + + // Warm the cache with the state before the click. + before, doc := f.do(t, "GET", "/api/toolfilter?tenant=me", "", jar) + if before.Code != http.StatusOK { + t.Fatalf("GET /api/toolfilter = %d %s", before.Code, before.Body) + } + if hasExcluded(doc, "Workflow") { + t.Fatalf("precondition: Workflow is already excluded: %v", doc) + } + if again, _ := f.do(t, "GET", "/api/toolfilter?tenant=me", "", jar); again.Header().Get("X-Cache") != "hit" { + t.Fatalf("second read X-Cache = %q; the test needs a warm cache to prove anything", + again.Header().Get("X-Cache")) + } + + if w, _ := f.do(t, "POST", "/api/toolfilter", + `{"kind":"tool","name":"Workflow","action":"exclude"}`, jar); w.Code != http.StatusOK { + t.Fatalf("exclude = %d %s", w.Code, w.Body) + } + after, doc := f.do(t, "GET", "/api/toolfilter?tenant=me", "", jar) + if !hasExcluded(doc, "Workflow") { + t.Errorf("the read right after the write still shows the old switch (X-Cache %q): %v", + after.Header().Get("X-Cache"), doc) + } + if got := after.Header().Get("X-Cache"); got == "stale" { + t.Errorf("served a stale body after a write that changed it") + } +} diff --git a/tenant/campaign.go b/tenant/campaign.go index 5d8140b5..8010b627 100644 --- a/tenant/campaign.go +++ b/tenant/campaign.go @@ -88,7 +88,7 @@ var ErrNoCampaign = errors.New("tenant: no such strategy campaign") // produce. The caller (proxy/campaign.go) has already resolved every cell's arm, // created the underlying keepalive_strategies rows, and validated the suggest payload // this came from — this method only persists the result. -func (r *Registry) CreateCampaign(actorID string, c Campaign, cells []CampaignCell) (Campaign, error) { +func (r *Registry) createCampaign(actorID string, c Campaign, cells []CampaignCell) (Campaign, error) { if c.Name == "" { return Campaign{}, fmt.Errorf("tenant: a campaign needs a name") } @@ -192,7 +192,7 @@ func (r *Registry) CampaignByID(id string) (Campaign, error) { // created — those stop being managed as a group, but they keep matching live traffic // until someone deletes or deactivates them individually through the strategy routes, // exactly as DeleteStrategy does not retroactively un-ping anything. -func (r *Registry) ArchiveCampaign(id string) error { +func (r *Registry) archiveCampaign(id string) error { res, err := r.db.Exec(`UPDATE strategy_campaigns SET status = ? WHERE id = ?`, CampaignStatusArchived, id) if err != nil { @@ -310,3 +310,20 @@ func nullableString(s string) any { } return s } + +// CreateCampaign and ArchiveCampaign commit and then tell the registry's listener (see SetOnChange). +func (r *Registry) CreateCampaign(actorID string, c Campaign, cells []CampaignCell) (Campaign, error) { + out, err := r.createCampaign(actorID, c, cells) + if err == nil { + r.changed(ChangeStrategy) + } + return out, err +} + +func (r *Registry) ArchiveCampaign(id string) error { + err := r.archiveCampaign(id) + if err == nil { + r.changed(ChangeStrategy) + } + return err +} diff --git a/tenant/keepalivestrategy.go b/tenant/keepalivestrategy.go index 27c4cd1e..fd01300a 100644 --- a/tenant/keepalivestrategy.go +++ b/tenant/keepalivestrategy.go @@ -303,7 +303,7 @@ func (s Strategy) Matches(now time.Time, tenantID string) bool { // caller (proxy's control route) validates the numeric bounds — the keep-alive's own // spend-authorization limits, which this package does not know about — and every // window and the target before calling this. -func (r *Registry) CreateStrategy(actorID string, s Strategy) (Strategy, error) { +func (r *Registry) createStrategy(actorID string, s Strategy) (Strategy, error) { if strings.TrimSpace(s.Name) == "" { return Strategy{}, fmt.Errorf("tenant: a strategy needs a name") } @@ -334,7 +334,7 @@ func (r *Registry) CreateStrategy(actorID string, s Strategy) (Strategy, error) } // UpdateStrategy applies a sparse patch and returns the resolved strategy. -func (r *Registry) UpdateStrategy(actorID, id string, p StrategyPatch) (Strategy, error) { +func (r *Registry) updateStrategy(actorID, id string, p StrategyPatch) (Strategy, error) { s, err := r.StrategyByID(id) if err != nil { return Strategy{}, err @@ -402,7 +402,7 @@ func (r *Registry) UpdateStrategy(actorID, id string, p StrategyPatch) (Strategy // DeleteStrategy removes a strategy. It does not un-ping anything already sent — there // is nothing to undo — it only stops the strategy matching new requests. -func (r *Registry) DeleteStrategy(id string) error { +func (r *Registry) deleteStrategy(id string) error { res, err := r.db.Exec(`DELETE FROM keepalive_strategies WHERE id = ?`, id) if err != nil { return err @@ -472,3 +472,29 @@ func scanStrategy(sc scanner) (Strategy, error) { out.CreatedAt, out.UpdatedAt = msTime(createdAt), msTime(updatedAt) return out, nil } + +// The three writers below commit through the unexported versions above and then tell the registry's +// listener, so a cached dashboard read that includes strategies is dropped the moment one changes. +func (r *Registry) CreateStrategy(actorID string, s Strategy) (Strategy, error) { + out, err := r.createStrategy(actorID, s) + if err == nil { + r.changed(ChangeStrategy) + } + return out, err +} + +func (r *Registry) UpdateStrategy(actorID, id string, p StrategyPatch) (Strategy, error) { + out, err := r.updateStrategy(actorID, id, p) + if err == nil { + r.changed(ChangeStrategy) + } + return out, err +} + +func (r *Registry) DeleteStrategy(id string) error { + err := r.deleteStrategy(id) + if err == nil { + r.changed(ChangeStrategy) + } + return err +} diff --git a/tenant/onchange_test.go b/tenant/onchange_test.go new file mode 100644 index 00000000..1b8e7f4a --- /dev/null +++ b/tenant/onchange_test.go @@ -0,0 +1,42 @@ +package tenant + +import "testing" + +// The dashboard drops cached reads that include account state when this fires, so it has to fire for +// every committed write to an account and for the strategy ledger, and not for a no-op. +func TestSetOnChangeFiresAfterCommittedWrites(t *testing.T) { + r, err := Open("", Options{}) + if err != nil { + t.Fatal(err) + } + var kinds []string + r.SetOnChange(func(kind string) { kinds = append(kinds, kind) }) + + tn, _, err := r.Register("laptop", "a@example.test") + if err != nil { + t.Fatal(err) + } + cfg := "pipeline: [format]\n" + if err := r.Update(tn, tn.ID, Patch{ConfigYAML: &cfg}); err != nil { + t.Fatal(err) + } + if len(kinds) != 1 || kinds[0] != ChangeAccount { + t.Fatalf("after an account update the listener saw %v, want [account]", kinds) + } + // The same document again changes nothing, so nothing is dropped. + if err := r.Update(tn, tn.ID, Patch{ConfigYAML: &cfg}); err != nil { + t.Fatal(err) + } + if len(kinds) != 1 { + t.Errorf("a no-op update fired the listener: %v", kinds) + } + // A refused write does not fire it either. + other := *tn + other.ID = "someone-else" + if err := r.Update(&other, tn.ID, Patch{ConfigYAML: &cfg}); err == nil { + t.Fatal("expected a forbidden update") + } + if len(kinds) != 1 { + t.Errorf("a refused update fired the listener: %v", kinds) + } +} diff --git a/tenant/tenant.go b/tenant/tenant.go index 5e78319a..9944c244 100644 --- a/tenant/tenant.go +++ b/tenant/tenant.go @@ -36,6 +36,7 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "time" _ "modernc.org/sqlite" // pure-Go driver, same as dash: no extra C toolchain @@ -278,6 +279,26 @@ type Registry struct { mu sync.RWMutex cache map[string]cacheEntry // sha256(token) hex -> resolved tenant + + onChange atomic.Pointer[func(kind string)] +} + +// SetOnChange registers fn to run after every committed write to an account (its configuration, its +// removal list, its upstreams, its label) and to the keep-alive strategy and campaign ledgers. The +// dashboard uses it to drop cached reads that include that state; without it a cached body would show +// the old state for as long as the cache may serve it. Set once at start-up. +func (r *Registry) SetOnChange(fn func(kind string)) { r.onChange.Store(&fn) } + +// ChangeAccount and ChangeStrategy are the kinds passed to the listener. +const ( + ChangeAccount = "account" + ChangeStrategy = "strategy" +) + +func (r *Registry) changed(kind string) { + if fn := r.onChange.Load(); fn != nil && *fn != nil { + (*fn)(kind) + } } type cacheEntry struct { @@ -767,6 +788,7 @@ func (r *Registry) Update(actor *Tenant, targetID string, p Patch) error { // hot, so precise invalidation buys nothing and can be wrong in one direction // that matters (a disabled tenant still being served). r.clearCache() + r.changed(ChangeAccount) // Disabling an account must also close its dashboard logins. Blocking the proxy // while leaving a live cookie behind would let a disabled user keep reading their // history for up to the session TTL, which is not what "disabled" means to