From 2200e46d985fe69cdcc892053da5fe969b8ef34e Mon Sep 17 00:00:00 2001 From: Jakub Zych Date: Wed, 4 Nov 2026 23:04:01 +0100 Subject: [PATCH] fix: rebuilds collect under one lock and keep moved lastmods A rebuild takes the transaction-scoped advisory lock golem15.sitemap.rebuild before it reads the settings and collects, and applies in the same transaction, so a second rebuild collects only after the first has committed. apply no longer writes the lastmod of a content row it did not create, so a lastmod RecordSaved moved meanwhile stays. --- README.md | 4 +- classes/rebuild.go | 79 ++++++++++++++++-------- filter_fixture_test.go | 20 +++++- rebuild_lock_test.go | 134 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 209 insertions(+), 28 deletions(-) create mode 100644 rebuild_lock_test.go diff --git a/README.md b/README.md index b0f5e00..5a8db5e 100644 --- a/README.md +++ b/README.md @@ -108,7 +108,7 @@ A `pact.RecordDeleted` takes the record's addresses out of the files at once. Bo ### Rebuilds -`classes.Rebuild` walks `signpost.Registry.ContentTypes` in registration order, pages `Records` 500 at a time, keeps the published records' locales with an address, asks the Filter per page, adds the additional addresses and writes everything in one transaction that holds the settings row (two rebuilds never interleave). It returns the added, changed and removed absolute URLs. +`classes.Rebuild` walks `signpost.Registry.ContentTypes` in registration order, pages `Records` 500 at a time, keeps the published records' locales with an address, asks the Filter per page, adds the additional addresses and writes everything. The whole rebuild runs in one transaction that first takes the advisory lock `golem15.sitemap.rebuild` and then reads the settings, collects and applies holding the settings row, so a second rebuild collects only after the first has committed and an older snapshot never overwrites a newer one. A rebuild never writes the lastmod of a content row it did not create (only an additional address takes its lastmod from the settings), so a lastmod a record save moved meanwhile stays. It returns the added, changed and removed absolute URLs. Record events set `rebuild_pending_until` with a conditional update and enqueue one rebuild with that delay only when no pending time lies in the future; the job clears it when it starts. A settings save dispatches a rebuild at once (a `summer_jobs` row labelled "Rebuild the sitemaps"). @@ -241,7 +241,7 @@ The files table lists the image and News files too. A save refuses an invalid si | `contract.Filter`, `contract.Candidate`, `contract.Decision`, `contract.Group` | The filter other plugins implement. | | `contract.ReasonNoindex`, `contract.ReasonCanonical`, `contract.ReasonRedirect`, `contract.ReasonType` | The left-out reasons. | | `contract.Listing`, `contract.ListingSettings`, `contract.GroupInfo`, `contract.EntriesQuery`, `contract.Entry`, `contract.EntriesPage` | What the sitemap publishes for other plugins. | -| `classes.Rebuild`, `classes.RebuildResult` | One rebuild of the slot table and what it changed. | +| `classes.Rebuild`, `classes.RebuildResult`, `classes.WithAfterSlotRead` | One rebuild of the slot table and what it changed; the test seam called after the rebuild read the slot rows. | | `classes.RebuildArgs`, `classes.RebuildKind` | The rebuild job's arguments and kind `golem15.sitemap.rebuild`. | | `classes.RequestRebuild`, `classes.DispatchRebuild`, `classes.RequestDebouncedRebuild`, `classes.ClearPending`, `classes.Building` | Dispatching rebuilds and the debounce window. | | `classes.RecordSaved`, `classes.RecordDeleted`, `classes.RealChanges`, `classes.Changed` | The record event handling. | diff --git a/classes/rebuild.go b/classes/rebuild.go index 47b21ca..6d24d37 100644 --- a/classes/rebuild.go +++ b/classes/rebuild.go @@ -114,14 +114,21 @@ type collected struct { // Rebuild reads every content type's published records (signpost Records, // D-93) and the additional addresses, writes them into the slot table and -// numbers the files, in one transaction that holds the settings row, so -// two rebuilds never interleave. A record that is gone or unpublished -// leaves its rows in the table with a reason, as a hole; only a changed -// file size renumbers the files (D-30). In production, with IndexNow on, -// the added, changed and removed addresses are queued for IndexNow after -// the commit (NotifyIndexNow), with the rows the record listeners marked +// numbers the files. A record that is gone or unpublished leaves its rows +// in the table with a reason, as a hole; only a changed file size +// renumbers the files (D-30). In production, with IndexNow on, the added, +// changed and removed addresses are queued for IndexNow after the commit +// (NotifyIndexNow), with the rows the record listeners marked // (indexnow_pending_at): a deleted record's addresses as removed, an // edited record's listed addresses as changed. +// +// Two rebuilds never interleave (L2 WR-02): one transaction takes the +// transaction-scoped advisory lock golem15.sitemap.rebuild first, then +// reads the settings and the type settings, collects the records, locks +// the settings row and applies, so a second rebuild collects only after +// the first has committed and an older snapshot never overwrites a newer +// one. A rebuild never writes the lastmod of a content row it did not +// create (WR-03), so a lastmod that RecordSaved moved meanwhile stays. func Rebuild(ctx context.Context, app *backpack.App) (RebuildResult, error) { var res RebuildResult if app == nil { @@ -132,22 +139,25 @@ func Rebuild(ctx context.Context, app *backpack.App) (RebuildResult, error) { return res, ErrNoDatabase } db = db.WithContext(ctx) - settings, err := LoadSettings(db) - if err != nil { - return res, err - } - types, err := LoadTypeSettings(db) - if err != nil { - return res, err - } reg, _ := app.Lookup[*signpost.Registry]() filter, _ := app.Lookup[contract.Filter]() - site := SiteURL(app, settings) - found, err := collect(ctx, reg, filter, settings, types, site) - if err != nil { - return res, err - } - err = db.Transaction(func(tx *gorm.DB) error { + err := db.Transaction(func(tx *gorm.DB) error { + if err := tx.Exec(`SELECT pg_advisory_xact_lock(hashtext('golem15.sitemap.rebuild'))`).Error; err != nil { + return err + } + settings, err := LoadSettings(tx) + if err != nil { + return err + } + types, err := LoadTypeSettings(tx) + if err != nil { + return err + } + site := SiteURL(app, settings) + found, err := collect(ctx, reg, filter, settings, types, site) + if err != nil { + return err + } if err := tx.Exec(`SELECT id FROM golem15_sitemap_settings WHERE id = 1 FOR UPDATE`).Error; err != nil { return err } @@ -155,7 +165,7 @@ func Rebuild(ctx context.Context, app *backpack.App) (RebuildResult, error) { if err != nil { return err } - res, err = apply(tx, found, locked, site) + res, err = apply(ctx, tx, found, locked, site) return err }) if err != nil { @@ -169,6 +179,16 @@ func Rebuild(ctx context.Context, app *backpack.App) (RebuildResult, error) { return res, nil } +// afterSlotReadKey carries the WithAfterSlotRead hook. +type afterSlotReadKey struct{} + +// WithAfterSlotRead returns ctx with fn, which a rebuild on ctx calls right +// after apply has read the slot rows and before it writes them: a test +// seam for what happens concurrently (nil, the default, calls nothing). +func WithAfterSlotRead(ctx context.Context, fn func()) context.Context { + return context.WithValue(ctx, afterSlotReadKey{}, fn) +} + // collect walks the content types page by page and adds the additional // addresses. The records of a type switched off in the settings are kept // in the table with the reason contract.ReasonType, so the settings screen @@ -411,12 +431,15 @@ func setAlternates(rows []wanted, defaultLocale string) { // apply writes the found addresses into the slot table and numbers the // files, inside the rebuild's transaction. -func apply(tx *gorm.DB, found collected, settings models.Settings, site string) (RebuildResult, error) { +func apply(ctx context.Context, tx *gorm.DB, found collected, settings models.Settings, site string) (RebuildResult, error) { var res RebuildResult var existing []models.URL if err := tx.Order("id ASC").Find(&existing).Error; err != nil { return res, err } + if fn, _ := ctx.Value(afterSlotReadKey{}).(func()); fn != nil { + fn() + } now := time.Now().UTC().Truncate(time.Microsecond) byKey := make(map[slotKey]*models.URL, len(existing)) for i := range existing { @@ -582,12 +605,11 @@ func apply(tx *gorm.DB, found collected, settings models.Settings, site string) for id := range dirty { r := rowsByID[id] r.UpdatedAt = now - if err := tx.Model(&models.URL{}).Where("id = ?", id).Updates(map[string]any{ + values := map[string]any{ "grp": r.Grp, "url": r.URL, "title": r.Title, "file_no": r.FileNo, - "lastmod": r.LastMod, "alternates": r.Alternates, "images": r.Images, "image_file_no": r.ImageFileNo, @@ -596,7 +618,14 @@ func apply(tx *gorm.DB, found collected, settings models.Settings, site string) "parent_refs": r.ParentRefs, "excluded_reason": r.ExcludedReason, "updated_at": r.UpdatedAt, - }).Error; err != nil { + } + // Only an additional address takes its lastmod from the rebuild; a + // content row's lastmod is RecordSaved's, never written back from + // this rebuild's read (WR-03). + if r.TypeKey == models.ExtraTypeKey { + values["lastmod"] = r.LastMod + } + if err := tx.Model(&models.URL{}).Where("id = ?", id).Updates(values).Error; err != nil { return res, err } } diff --git a/filter_fixture_test.go b/filter_fixture_test.go index bbe3b19..72ab4e3 100644 --- a/filter_fixture_test.go +++ b/filter_fixture_test.go @@ -16,16 +16,34 @@ const filterPluginID = "acme.filter" // fakeFilter excludes addresses containing "noindex" (reason noindex) and // puts addresses containing "promo" into the group offers. It records the -// batches it was asked. +// batches it was asked; hook, when set, runs at the start of every Decide +// (a test blocks a rebuild there). type fakeFilter struct { mu sync.Mutex batches [][]contract.Candidate + hook func() +} + +// setHook sets the Decide hook for one test. +func (f *fakeFilter) setHook(t interface{ Cleanup(func()) }, fn func()) { + f.mu.Lock() + f.hook = fn + f.mu.Unlock() + t.Cleanup(func() { + f.mu.Lock() + f.hook = nil + f.mu.Unlock() + }) } func (f *fakeFilter) Decide(_ context.Context, batch []contract.Candidate) ([]contract.Decision, error) { f.mu.Lock() f.batches = append(f.batches, append([]contract.Candidate(nil), batch...)) + hook := f.hook f.mu.Unlock() + if hook != nil { + hook() + } out := make([]contract.Decision, len(batch)) for i, c := range batch { switch { diff --git a/rebuild_lock_test.go b/rebuild_lock_test.go new file mode 100644 index 0000000..35a381d --- /dev/null +++ b/rebuild_lock_test.go @@ -0,0 +1,134 @@ +package sitemap_test + +import ( + "context" + "slices" + "sync" + "testing" + "time" + + "git.golem15.com/golem15/sm-sitemap-plugin/classes" + "git.golem15.com/golem15/summercms/modules/pact" +) + +// rebuildResult is the outcome of one classes.Rebuild run in a goroutine. +type rebuildResult struct { + res classes.RebuildResult + err error +} + +// waitFor polls cond until it holds or the deadline passes. +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(15 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(20 * time.Millisecond) + } +} + +// TestConcurrentRebuilds pins L2 WR-02 (D-30, P33-63): a second rebuild +// waits on the rebuild's advisory lock while the first one collects, so it +// collects only after the first has committed; a record published while +// the first rebuild ran is listed afterwards and never sent to IndexNow as +// removed. +func TestConcurrentRebuilds(t *testing.T) { + e := bootIndex(t, filterPluginID) + mod := time.Date(2026, 10, 1, 8, 0, 0, 0, time.UTC) + contentStore.put(pageType, page1(1, "a", mod)) + if _, err := classes.Rebuild(t.Context(), e.app); err != nil { + t.Fatal(err) + } + + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + theFilter.setHook(t, func() { + first := false + once.Do(func() { first = true; close(entered) }) + if first { + <-release + } + }) + run := func() <-chan rebuildResult { + out := make(chan rebuildResult, 1) + go func() { + res, err := classes.Rebuild(context.Background(), e.app) + out <- rebuildResult{res, err} + }() + return out + } + a := run() + select { + case <-entered: + case <-time.After(15 * time.Second): + t.Fatal("rebuild A never asked the filter") + } + contentStore.put(pageType, page1(2, "x", mod)) + b := run() + waitFor(t, "rebuild B to wait on the advisory lock", func() bool { + var waiting int64 + e.gdb.Raw(`SELECT COUNT(*) FROM pg_locks WHERE locktype = 'advisory' AND NOT granted`).Scan(&waiting) + return waiting > 0 + }) + close(release) + ra, rb := <-a, <-b + if ra.err != nil || rb.err != nil { + t.Fatalf("rebuilds: %v, %v", ra.err, rb.err) + } + var reason []string + e.gdb.Raw(`SELECT excluded_reason FROM golem15_sitemap_urls WHERE type_key = ? AND record_id = '2'`, pageType).Scan(&reason) + if len(reason) != 1 || reason[0] != "" { + t.Fatalf("record x after both rebuilds = %v, want listed", reason) + } + x := "https://api.example.test/pages/x" + if slices.Contains(ra.res.Removed, x) || slices.Contains(rb.res.Removed, x) { + t.Fatalf("x sent as removed: A %v, B %v", ra.res.Removed, rb.res.Removed) + } + if !slices.Contains(rb.res.Added, x) { + t.Fatalf("rebuild B added %v, want x", rb.res.Added) + } +} + +// TestRebuildKeepsMovedLastmod pins L2 WR-03 (D-30): a lastmod that +// RecordSaved moves while a rebuild runs stays, also when the same rebuild +// rewrites the row for another change (its title). +func TestRebuildKeepsMovedLastmod(t *testing.T) { + e := bootIndex(t) + old := time.Date(2026, 10, 1, 8, 0, 0, 0, time.UTC) + contentStore.put(pageType, page1(1, "a", old)) + if _, err := classes.Rebuild(t.Context(), e.app); err != nil { + t.Fatal(err) + } + renamed := page1(1, "a", old) + renamed.Locales[0].Title = "Page a, renamed" + contentStore.put(pageType, renamed) + + lastmod := func() time.Time { + var at time.Time + e.gdb.Raw(`SELECT lastmod FROM golem15_sitemap_urls WHERE type_key = ? AND record_id = '1'`, pageType).Scan(&at) + return at + } + saved := make(chan error, 1) + ctx := classes.WithAfterSlotRead(t.Context(), func() { + go func() { + saved <- e.app.Events.Fire(context.Background(), &pact.RecordSaved{Model: pageModel, ID: "1", Changes: []pact.RecordChange{{Attribute: "title"}}}) + }() + waitFor(t, "RecordSaved to move the lastmod", func() bool { return !lastmod().Equal(old) }) + }) + if _, err := classes.Rebuild(ctx, e.app); err != nil { + t.Fatal(err) + } + if err := <-saved; err != nil { + t.Fatal(err) + } + if got := lastmod(); got.Equal(old) { + t.Fatal("the rebuild wrote the lastmod it read back over the moved one") + } + var title string + e.gdb.Raw(`SELECT title FROM golem15_sitemap_urls WHERE type_key = ? AND record_id = '1'`, pageType).Scan(&title) + if title != "Page a, renamed" { + t.Fatalf("title = %q, want the rebuild's change", title) + } +}