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
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user