From db6a7f5c3f192feb4b8527cca233e19503a896f9 Mon Sep 17 00:00:00 2001 From: Jakub Zych Date: Thu, 15 Oct 2026 22:36:24 +0200 Subject: [PATCH] fix(19.3-06): queue site-cache purges after commit and coalesce them (WR-01) - PurgeSiteCacheAfterCommit queues the purge on a per-purger worker after commit: one request in flight, saves committed meanwhile collapse into one follow-up, so the admin request never waits for the site - new cooler.SiteCachePurger.Wait drains the queue; Handle.Close waits for it at shutdown (timeout plus one second) - Purge stays the synchronous send; exported names and signatures unchanged --- docs/services/caching.md | 18 ++-- modules/cooler/README.md | 9 +- modules/cooler/cooler.go | 27 ++++-- modules/cooler/example_test.go | 10 +- modules/cooler/sitecache.go | 155 ++++++++++++++++++++++++++----- modules/cooler/sitecache_test.go | 65 ++++++++++++- 6 files changed, 236 insertions(+), 48 deletions(-) diff --git a/docs/services/caching.md b/docs/services/caching.md index 295dc39..1071bf5 100644 --- a/docs/services/caching.md +++ b/docs/services/caching.md @@ -97,7 +97,7 @@ The `serve` command installs cooler at start-up (`cooler.Install`) and publishes ## Site cache purge -A server-rendered site in front of the API usually caches its pages for a few minutes (a Nuxt `routeRules` `swr` window, for example). Without help, an edit in the admin shows on the site only when that window ends. cooler closes the gap: after an admin save commits, it sends the site one purge request, and the site drops its route cache, so the next page view renders the change. +A server-rendered site in front of the API usually caches its pages for a few minutes (a Nuxt `routeRules` `swr` window, for example). Without help, an edit in the admin shows on the site only when that window ends. cooler closes the gap: after an admin save commits, it queues a purge request to the site, and the site drops its route cache, so the next page view renders the change. Configure the target once for the application: @@ -110,7 +110,7 @@ Set both or neither; with neither, nothing is purged. The URL must be http or ht ### What a purge sends -Each committed save sends one request, with no debounce: +Each committed save queues a purge; the request looks like this: ```http POST /api/__site__/cache/purge @@ -122,20 +122,24 @@ Content-Type: application/json The kind names what was saved (a plugin picks it, for example `faq`, `menu`, `menu_entry`, `post` or `category`) and the slug which record; the site may log them. The site answers `{"purged": }`, and cooler logs `cooler: site cache purged` with the kind, slug, status and count. +At most one request is in flight per process (per purger). Saves committed while it is sent collapse into one follow-up request carrying the last kind and slug, because each purge clears the site's whole route cache: a bulk delete of fifty rows sends at most two requests. The log line of such a follow-up adds `coalesced` with the number of saves it stands in for besides its own. + ### Timing and failures -`cooler.PurgeSiteCacheAfterCommit` registers the request on the save's transaction through `lagoon.AfterCommit`. It is sent only after the commit, so a page view between the purge and the commit cannot re-cache the old content, and a save that fails or is rolled back sends nothing. Outside a transaction it is sent at once. +`cooler.PurgeSiteCacheAfterCommit` registers the purge on the save's transaction through `lagoon.AfterCommit`. It is queued only after the commit, so a page view between the purge and the commit cannot re-cache the old content, and a save that fails or is rolled back sends nothing. Outside a transaction it is queued at once. Queuing never blocks: the admin request answers without waiting for the site, and a background worker sends the request. -A purge never fails a save. The request is bounded by `cooler.SiteCachePurgeTimeout` (2 s) and detached from the admin request's cancellation; an unreachable site, an error status or a timeout is logged as one `cooler: site cache purge failed` warning and the save answers as usual. Redirects are not followed: a 3xx answer is logged as a failed purge with its status, so the bearer token never goes to another URL and the POST never turns into a GET. Point `site_cache.purge_url` at the final URL. +A purge never fails a save. Each request is bounded by `cooler.SiteCachePurgeTimeout` (2 s) and detached from the admin request's cancellation; an unreachable site, an error status or a timeout is logged as one `cooler: site cache purge failed` warning and the save answers as usual. Redirects are not followed: a 3xx answer is logged as a failed purge with its status, so the bearer token never goes to another URL and the POST never turns into a GET. Point `site_cache.purge_url` at the final URL. + +Because the purge is fire and forget, a test or a console command that wants to see its effect calls `cooler.SiteCachePurger.Wait`, which returns once nothing is queued or in flight (or when its context ends). `serve` waits for a queued purge on shutdown, up to the timeout plus one second, through `cooler.Handle.Close`. ### Purging from a plugin The `serve` command publishes the purger at start-up (`cooler.Install`), and `cooler.SiteCacheFromApp` returns it, or builds it from the keys in commands and tests that do not run `serve`. An admin controller calls it from its after-save hooks, which run inside the save transaction, and takes that transaction from `cabana.TxFromContext`: ```go src=modules/cooler/example_test.go#articlesController.purge -// purge registers one site-cache purge for the saved article, sent after the -// save transaction commits (at once outside one). Without a purge configured -// the purger is nil and nothing is sent. +// purge queues one site-cache purge for the saved article once the save +// transaction commits (at once outside one); the save never waits for the +// site. Without a purge configured the purger is nil and nothing is sent. func (c articlesController) purge(ctx context.Context, model any) error { article, ok := model.(*Article) if !ok { diff --git a/modules/cooler/README.md b/modules/cooler/README.md index 5361a39..e9deef2 100644 --- a/modules/cooler/README.md +++ b/modules/cooler/README.md @@ -36,7 +36,7 @@ Use cases: - Keys carry a per-application prefix: `cache.prefix`, or by default `app.name` lower-cased with every run of other characters than `a-z` and `0-9` replaced by `_`, then `:` (`summer:` when `app.name` is empty). Cache keys are `cache:`, rate-limit keys `rl:`. - `cooler.OpenRedis` accepts `redis://`, `rediss://` (TLS) and `unix://` URLs, does not dial at construction, and skips the `CLIENT SETINFO` and maintenance-notification handshakes. - Only `serve` installs cooler: `route:list` and `surf.Assemble` in tests keep the in-process store and never dial Redis. -- Site-cache purge: `cooler.SiteCachePurger` sends `POST {"kind":…,"slug":…}` with `Authorization: Bearer ` to `site_cache.purge_url`. `cooler.PurgeSiteCacheAfterCommit` registers that request on the save transaction through `lagoon.AfterCommit`, so it runs only after the save commits and never for a save that rolled back. One request per committed save, no debounce. +- Site-cache purge: `cooler.SiteCachePurger` sends `POST {"kind":…,"slug":…}` with `Authorization: Bearer ` to `site_cache.purge_url`. `cooler.PurgeSiteCacheAfterCommit` queues that request on the save transaction through `lagoon.AfterCommit`, so it is queued only after the save commits and never for a save that rolled back. The save never waits for the site: a per-purger worker sends the request, with at most one in flight, and saves committed meanwhile collapse into one follow-up request (each purge clears the site's whole route cache), so a bulk delete sends at most two. `cooler.SiteCachePurger.Wait` drains the queue for tests and commands, and `cooler.Handle.Close` drains it at shutdown. - A purge never fails a save: each request is bounded by `cooler.SiteCachePurgeTimeout` (2 s) and detached from the request's cancellation, and a failure, an error status or a timeout is one warning (`cooler: site cache purge failed`) with the kind, slug and status. A success is one info line (`cooler: site cache purged`) with the purged count the site answered. Log lines and errors never carry the token or the full URL. - The purge URL must be http or https with a host and without user info or a fragment; plain http is accepted only for a loopback host (127.0.0.0/8, `::1`, `localhost`), so the token never crosses a network in clear. The token needs at least `cooler.SiteCachePurgeMinTokenLength` (32) characters, the same minimum the site's route applies. Setting only one of the two keys, or an invalid value, fails `serve` at boot with an error that names the key. - `cooler.Install` publishes the purger at boot unless a plugin already published one (a plugin that still reads its own deprecated keys), and `cooler.SiteCacheFromApp` gives admin hooks the same purger lazily in commands and tests that do not run `serve`. Plugins take the save transaction from `cabana.TxFromContext` themselves; cooler imports neither surf nor cabana. @@ -164,7 +164,7 @@ func AllowExport(ctx context.Context, app *backpack.App, fallback cooler.Attempt | `cooler.Install` | Reads cooler's config, publishes the Cache, opens and pings Redis when a Redis driver is selected, and returns a `cooler.Handle`. Called by surf's `serve`. | | `cooler.Handle` | What `cooler.Install` set up. | | `cooler.Handle.Cache` | The Cache `cooler.Install` published or found. | -| `cooler.Handle.Close` | Closes the Redis client `cooler.Install` opened, if any. | +| `cooler.Handle.Close` | Waits for the site-cache purges queued on the purger `cooler.Install` published (up to `cooler.SiteCachePurgeTimeout` plus one second), then closes the Redis client `cooler.Install` opened, if any. | | `cooler.Handle.RateLimitRedis` | Reports whether `http.rate_limit.store` selects the Redis store. | | `cooler.Handle.RateLimitStore` | Returns the Redis rate-limit store with the given fallback, or the fallback itself when Redis is not selected. | | `cooler.Attempter` | One atomic fixed-window admission; the method set of `surf.Store`. | @@ -175,9 +175,10 @@ func AllowExport(ctx context.Context, app *backpack.App, fallback cooler.Attempt | `cooler.SiteCachePurger` | Sends the site-cache purge request; a nil purger does nothing. | | `cooler.NewSiteCachePurger` | Builds the purger from the values of `site_cache.purge_url` and `site_cache.purge_token`; both empty returns nil; errors name those keys and never the token. | | `cooler.NewSiteCachePurgerFor` | The same with the config keys its errors name passed in, for a plugin that reads deprecated keys of its own. | -| `cooler.SiteCachePurger.Purge` | Sends one purge request (2 s timeout, detached from cancellation, redirects not followed) and logs the outcome; never fails. | +| `cooler.SiteCachePurger.Purge` | Sends one purge request and blocks until it is done (2 s timeout, detached from cancellation, redirects not followed); the send the after-commit queue uses. Logs the outcome; never fails. | +| `cooler.SiteCachePurger.Wait` | Blocks until no queued purge is pending or in flight, or until the context ends (returning its error); nil at once for a nil or idle purger. | | `cooler.SiteCachePurger.Endpoint` | The purge URL's host and path, for log lines. | -| `cooler.PurgeSiteCacheAfterCommit` | Registers one purge to run after the given transaction commits (at once outside a transaction). | +| `cooler.PurgeSiteCacheAfterCommit` | Queues one purge after the given transaction commits (at once outside a transaction), fire and forget: one request in flight per purger, later saves collapse into one follow-up. | | `cooler.SiteCacheFromApp` | Returns the published purger, or builds it from the framework keys and publishes it; nil when unset or invalid. | | `cooler.SiteCachePurgeURLKey` | `site_cache.purge_url`. | | `cooler.SiteCachePurgeTokenKey` | `site_cache.purge_token`. | diff --git a/modules/cooler/cooler.go b/modules/cooler/cooler.go index 16c13f3..d780ff1 100644 --- a/modules/cooler/cooler.go +++ b/modules/cooler/cooler.go @@ -94,21 +94,32 @@ type Cache interface { Flush(ctx context.Context) error } -// Handle is what Install set up. Close releases the Redis client Install -// opened, if any. +// Handle is what Install set up. Close drains the site-cache purger and +// releases the Redis client Install opened, if any. type Handle struct { client *redis.Client prefix string rateRedis bool cache Cache log *slog.Logger + purger *SiteCachePurger } -// Close closes the Redis client Install opened. It is safe on a nil Handle -// and when no Redis driver was selected. A Redis-backed Cache that Install -// built on that client stops working; serve closes the handle on shutdown. +// Close first waits for the site-cache purges queued on the purger Install +// published, up to SiteCachePurgeTimeout plus one second, then closes the +// Redis client Install opened. It is safe on a nil Handle and when no Redis +// driver was selected. A Redis-backed Cache that Install built on that +// client stops working; serve closes the handle on shutdown. func (h *Handle) Close() error { - if h == nil || h.client == nil { + if h == nil { + return nil + } + if h.purger != nil { + ctx, cancel := context.WithTimeout(context.Background(), SiteCachePurgeTimeout+time.Second) + _ = h.purger.Wait(ctx) // the deadline only bounds shutdown + cancel() + } + if h.client == nil { return nil } err := h.client.Close() @@ -169,9 +180,11 @@ func Install(ctx context.Context, app *backpack.App) (*Handle, error) { } cfg := app.Config h := &Handle{prefix: keyPrefix(cfg), log: loggerFromApp(app)} - if err := installSiteCache(app, h.log); err != nil { + purger, err := installSiteCache(app, h.log) + if err != nil { return nil, err } + h.purger = purger store, err := driverName(cfg, RateLimitStoreKey) if err != nil { return nil, err diff --git a/modules/cooler/example_test.go b/modules/cooler/example_test.go index de6727b..73e0daa 100644 --- a/modules/cooler/example_test.go +++ b/modules/cooler/example_test.go @@ -59,9 +59,9 @@ type articlesController struct { app func() *backpack.App } -// purge registers one site-cache purge for the saved article, sent after the -// save transaction commits (at once outside one). Without a purge configured -// the purger is nil and nothing is sent. +// purge queues one site-cache purge for the saved article once the save +// transaction commits (at once outside one); the save never waits for the +// site. Without a purge configured the purger is nil and nothing is sent. func (c articlesController) purge(ctx context.Context, model any) error { article, ok := model.(*Article) if !ok { @@ -121,5 +121,9 @@ func ExampleSiteCacheFromApp() { if err := c.FormAfterUpdate(context.Background(), &Article{Slug: "hello-summer"}); err != nil { fmt.Println(err) } + // The purge is queued; Wait drains the queue (serve does it at shutdown). + if err := cooler.SiteCacheFromApp(app).Wait(context.Background()); err != nil { + fmt.Println(err) + } // Output: POST /api/__site__/cache/purge {"kind":"article","slug":"hello-summer"} } diff --git a/modules/cooler/sitecache.go b/modules/cooler/sitecache.go index 517458f..7cf9d52 100644 --- a/modules/cooler/sitecache.go +++ b/modules/cooler/sitecache.go @@ -12,6 +12,7 @@ import ( "net/http" "net/url" "strings" + "sync" "time" "git.golem15.com/golem15/summercms/modules/backpack" @@ -40,11 +41,26 @@ const ( // SiteCachePurger asks the host site to drop its page caches after an admin // save. A nil *SiteCachePurger is disabled and does nothing. +// +// Purges queued by PurgeSiteCacheAfterCommit go through one worker per +// purger: at most one request is in flight, and the saves committed while +// it is sent collapse into one follow-up request carrying the last kind and +// slug (each purge clears the site's whole route cache). type SiteCachePurger struct { url string endpoint string token string client *http.Client + + mu sync.Mutex + // sending is true while the worker goroutine runs. + sending bool + // pending is the follow-up body, the latest one queued while sending. + pending *siteCachePurgeBody + // coalesced counts the queued bodies pending replaced. + coalesced int + // idle is closed when the current busy period ends; nil while idle. + idle chan struct{} } // NewSiteCachePurger returns the purger for purgeURL and token, read from @@ -122,44 +138,62 @@ type siteCachePurgeBody struct { // Purge sends the purge request for one saved entity: a POST of // {"kind":kind,"slug":slug} with the bearer token, bounded by -// SiteCachePurgeTimeout and detached from ctx's cancellation. It logs the -// outcome and returns nothing: a purge never fails a save. A redirect is -// never followed; a 3xx answer is logged as a failed purge with its status. -// The token is never logged. +// SiteCachePurgeTimeout and detached from ctx's cancellation. It blocks +// until the request is done; it is the send the PurgeSiteCacheAfterCommit +// queue uses. It logs the outcome and returns nothing: a purge never fails +// a save. A redirect is never followed; a 3xx answer is logged as a failed +// purge with its status. The token is never logged. func (p *SiteCachePurger) Purge(ctx context.Context, kind, slug string) { if p == nil { return } + p.send(ctx, kind, slug, 0) +} + +// send is Purge; coalesced, when above zero, is logged with the outcome: +// the number of queued purges this request stands in for besides its own. +func (p *SiteCachePurger) send(ctx context.Context, kind, slug string, coalesced int) { if ctx == nil { ctx = context.Background() } ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), SiteCachePurgeTimeout) defer cancel() log := slog.Default() + failed := func(attrs ...any) { + attrs = append([]any{"kind", kind, "slug", slug}, attrs...) + if coalesced > 0 { + attrs = append(attrs, "coalesced", coalesced) + } + log.Warn("cooler: site cache purge failed", attrs...) + } raw, err := json.Marshal(siteCachePurgeBody{Kind: kind, Slug: slug}) if err != nil { - log.Warn("cooler: site cache purge failed", "kind", kind, "slug", slug, "error", err.Error()) + failed("error", err.Error()) return } req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.url, bytes.NewReader(raw)) if err != nil { - log.Warn("cooler: site cache purge failed", "kind", kind, "slug", slug, "error", "invalid request") + failed("error", "invalid request") return } req.Header.Set("Authorization", "Bearer "+p.token) req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "application/json") - res, err := p.client.Do(req) + client := p.client + if client == nil { + client = &http.Client{CheckRedirect: noRedirect} + } + res, err := client.Do(req) if err != nil { - log.Warn("cooler: site cache purge failed", "kind", kind, "slug", slug, "error", purgeTransportError(err)) + failed("error", purgeTransportError(err)) return } defer res.Body.Close() body, _ := io.ReadAll(io.LimitReader(res.Body, siteCacheAnswerLimit)) if res.StatusCode < 200 || res.StatusCode > 299 { - log.Warn("cooler: site cache purge failed", "kind", kind, "slug", slug, "status", res.StatusCode) + failed("status", res.StatusCode) return } attrs := []any{"kind", kind, "slug", slug, "status", res.StatusCode} @@ -169,9 +203,79 @@ func (p *SiteCachePurger) Purge(ctx context.Context, kind, slug string) { if json.Unmarshal(body, &answer) == nil && answer.Purged != nil { attrs = append(attrs, "purged", *answer.Purged) } + if coalesced > 0 { + attrs = append(attrs, "coalesced", coalesced) + } log.Info("cooler: site cache purged", attrs...) } +// enqueue queues one purge. With nothing in flight it starts the worker, +// which sends at once; while a request is in flight the body becomes the +// pending follow-up, replacing (and counting) an earlier pending one. +func (p *SiteCachePurger) enqueue(kind, slug string) { + body := siteCachePurgeBody{Kind: kind, Slug: slug} + p.mu.Lock() + if p.sending { + if p.pending != nil { + p.coalesced++ + } + p.pending = &body + p.mu.Unlock() + return + } + p.sending = true + p.idle = make(chan struct{}) + p.mu.Unlock() + go p.work(body) +} + +// work sends body, then each pending follow-up, until none is left; then it +// ends the busy period. +func (p *SiteCachePurger) work(body siteCachePurgeBody) { + coalesced := 0 + for { + p.send(context.Background(), body.Kind, body.Slug, coalesced) + p.mu.Lock() + if p.pending == nil { + p.sending = false + close(p.idle) + p.idle = nil + p.mu.Unlock() + return + } + body, coalesced = *p.pending, p.coalesced + p.pending, p.coalesced = nil, 0 + p.mu.Unlock() + } +} + +// Wait blocks until no queued purge is pending or in flight, or until ctx +// ends, then returning ctx.Err(). It returns nil at once for a nil or idle +// purger. Tests and commands call it before they read what a purge did; +// cooler's Handle.Close calls it at shutdown. +func (p *SiteCachePurger) Wait(ctx context.Context) error { + if p == nil { + return nil + } + if ctx == nil { + ctx = context.Background() + } + for { + p.mu.Lock() + idle := p.idle + p.mu.Unlock() + if idle == nil { + return nil + } + select { + case <-idle: + // A new busy period may have started; look again. + case <-ctx.Done(): + return ctx.Err() + } + } +} + // purgeTransportError describes a failed request without the request URL // or headers. func purgeTransportError(err error) string { @@ -185,18 +289,22 @@ func purgeTransportError(err error) string { return err.Error() } -// PurgeSiteCacheAfterCommit registers p.Purge(kind, slug) to run after the +// PurgeSiteCacheAfterCommit queues a purge of kind and slug after the // transaction db belongs to commits, through lagoon.AfterCommit: a failed // save never purges, and a concurrent page view cannot re-cache the old -// content between the purge and the commit. Each call sends one request -// after its commit; there is no debounce. A nil p does nothing; outside a -// transaction (db nil or not in one) the purge runs at once. +// content between the purge and the commit. The after-commit callback only +// queues, so the save never waits for the site (fire and forget). At most +// one request is in flight per purger, and saves committed meanwhile +// collapse into one follow-up request with the last kind and slug, since +// each purge clears the site's whole route cache. A nil p does nothing; +// outside a transaction (db nil or not in one) the purge is queued at once. +// SiteCachePurger.Wait waits for the queue to drain. func PurgeSiteCacheAfterCommit(ctx context.Context, db *gorm.DB, p *SiteCachePurger, kind, slug string) { if p == nil { return } - lagoon.AfterCommit(ctx, db, func(ctx context.Context, _ *gorm.DB) { - p.Purge(ctx, kind, slug) + lagoon.AfterCommit(ctx, db, func(context.Context, *gorm.DB) { + p.enqueue(kind, slug) }) } @@ -244,23 +352,24 @@ func siteCacheFromConfig(app *backpack.App) (*SiteCachePurger, error) { } // installSiteCache publishes the site-cache purger unless one is already -// published (a plugin's alias configuration): an invalid configuration is a -// boot error naming the key. -func installSiteCache(app *backpack.App, log *slog.Logger) error { +// published (a plugin's alias configuration) and returns the purger in +// effect, nil when none is configured: an invalid configuration is a boot +// error naming the key. +func installSiteCache(app *backpack.App, log *slog.Logger) (*SiteCachePurger, error) { if p, ok := app.Lookup[*SiteCachePurger](); ok && p != nil { log.Info("cooler: site cache purge enabled", "endpoint", p.Endpoint()) - return nil + return p, nil } p, err := siteCacheFromConfig(app) if err != nil { - return err + return nil, err } if p == nil { - return nil + return nil, nil } if err := app.Publish(p); err != nil { - return fmt.Errorf("cooler: %w", err) + return nil, fmt.Errorf("cooler: %w", err) } log.Info("cooler: site cache purge enabled", "endpoint", p.Endpoint()) - return nil + return p, nil } diff --git a/modules/cooler/sitecache_test.go b/modules/cooler/sitecache_test.go index ce94928..18ce161 100644 --- a/modules/cooler/sitecache_test.go +++ b/modules/cooler/sitecache_test.go @@ -244,6 +244,16 @@ func newTestPurger(t *testing.T, srv *httptest.Server) *SiteCachePurger { return p } +// waitPurges drains the purges queued on p, failing the test after 5 s. +func waitPurges(t *testing.T, p *SiteCachePurger) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := p.Wait(ctx); err != nil { + t.Fatalf("Wait: %v", err) + } +} + func assertNoToken(t *testing.T, logs *syncBuffer) { t.Helper() for _, line := range logs.lines() { @@ -371,10 +381,12 @@ func TestSiteCachePurgeRequest(t *testing.T) { } }) - t.Run("outside a transaction the after-commit purge runs at once", func(t *testing.T) { + t.Run("outside a transaction the after-commit purge is queued at once", func(t *testing.T) { captureLogs(t) srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(`{"purged":2}`)) }) - PurgeSiteCacheAfterCommit(context.Background(), nil, newTestPurger(t, srv), "category", "now") + p := newTestPurger(t, srv) + PurgeSiteCacheAfterCommit(context.Background(), nil, p, "category", "now") + waitPurges(t, p) if got := hits(); len(got) != 1 || got[0].body["kind"] != "category" || got[0].body["slug"] != "now" { t.Fatalf("requests = %#v, want one", got) } @@ -464,6 +476,7 @@ func TestSiteCacheAfterCommitOnly(t *testing.T) { if n := countPosts(t, gdb, "rolled-back"); n != 0 { t.Fatalf("%d rows after rollback", n) } + waitPurges(t, p) if got := hits(); len(got) != 0 { t.Fatalf("a rolled-back save sent %d purges", len(got)) } @@ -485,6 +498,7 @@ func TestSiteCacheAfterCommitOnly(t *testing.T) { if during != 0 { t.Fatalf("%d purges were sent before the commit", during) } + waitPurges(t, p) got := hits() if len(got) != 1 || got[0].body["kind"] != "post" || got[0].body["slug"] != "committed" { t.Fatalf("purges = %#v, want one for post committed", got) @@ -510,6 +524,7 @@ func TestSiteCacheAfterCommitOnly(t *testing.T) { if afterInner != before { t.Fatal("the purge ran when the inner transaction finished") } + waitPurges(t, p) if got := hits(); len(got) != before+1 || got[len(got)-1].body["slug"] != "nested" { t.Fatalf("purges = %#v, want one more for nested", got) } @@ -531,6 +546,7 @@ func TestSiteCacheFailureNeverFails(t *testing.T) { if err != nil { t.Fatalf("a failing purge failed the save: %v", err) } + waitPurges(t, p) if n := countPosts(t, gdb, "server-error"); n != 1 || len(hits()) != 1 { t.Fatalf("rows %d, purges %d; want 1 and 1", n, len(hits())) } @@ -559,12 +575,13 @@ func TestSiteCacheFailureNeverFails(t *testing.T) { if err != nil { t.Fatalf("a slow purge failed the save: %v", err) } - if elapsed := time.Since(start); elapsed > SiteCachePurgeTimeout+2*time.Second { - t.Fatalf("the save waited %s, want the purge bounded by %s", elapsed, SiteCachePurgeTimeout) + if elapsed := time.Since(start); elapsed > 500*time.Millisecond { + t.Fatalf("the save waited %s for a queued purge", elapsed) } if n := countPosts(t, gdb, "slow"); n != 1 { t.Fatalf("%d rows, want the save committed", n) } + waitPurges(t, p) if lines := logs.lines(); len(lines) != 1 || !strings.Contains(lines[0], "error=timeout") { t.Fatalf("log = %q", lines) } @@ -582,6 +599,7 @@ func TestSiteCacheFailureNeverFails(t *testing.T) { if err != nil || countPosts(t, gdb, "unreachable") != 1 { t.Fatalf("Transaction = %v, rows %d", err, countPosts(t, gdb, "unreachable")) } + waitPurges(t, p) }) } @@ -660,3 +678,42 @@ func TestSiteCachePurgeNoRedirect(t *testing.T) { } assertNoToken(t, logs) } + +func TestSiteCachePurgeAsyncCoalesced(t *testing.T) { + logs := captureLogs(t) + release := make(chan struct{}) + srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { + <-release + _, _ = w.Write([]byte(`{"purged":3}`)) + }) + p := newTestPurger(t, srv) + released := false + defer func() { + if !released { + close(release) + } + waitPurges(t, p) + }() + for i := range 20 { + start := time.Now() + PurgeSiteCacheAfterCommit(context.Background(), nil, p, "post", "slug-"+strconv.Itoa(i)) + if elapsed := time.Since(start); elapsed > 50*time.Millisecond { + t.Fatalf("call %d blocked for %s", i, elapsed) + } + } + close(release) + released = true + waitPurges(t, p) + got := hits() + if len(got) != 2 { + t.Fatalf("%d requests, want 2 (one in flight, one coalesced follow-up)", len(got)) + } + if got[0].body["slug"] != "slug-0" || got[1].body["slug"] != "slug-19" { + t.Fatalf("slugs %q, %q; want slug-0 then the last, slug-19", got[0].body["slug"], got[1].body["slug"]) + } + lines := logs.lines() + if len(lines) != 2 || strings.Contains(lines[0], "coalesced=") || !strings.Contains(lines[1], "slug=slug-19") || !strings.Contains(lines[1], "coalesced=18") { + t.Fatalf("log = %q, want the second purged line with coalesced=18", lines) + } + assertNoToken(t, logs) +}