c71b58eea7
The store keeps FinishTimeout, the final status write and the conga job end get their own 30 s bound and prune its own store-sized bound, so the row no longer stays running with the error lost until RecoverStale; adds the code-only crawl.WithFinishTimeout seam. quick-261104-xae, review WR-03.
356 lines
12 KiB
Go
356 lines
12 KiB
Go
package seo_test
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.golem15.com/golem15/sm-seo-plugin/classes/crawl"
|
|
"git.golem15.com/golem15/sm-seo-plugin/models"
|
|
"git.golem15.com/golem15/summercms/modules/fetchguard"
|
|
)
|
|
|
|
// blockingSite is a fakeSite whose answer for one path waits until it is
|
|
// released or the request's context ends (then the request fails with the
|
|
// context's error, as a real transport does). reached is closed when that
|
|
// request arrives.
|
|
type blockingSite struct {
|
|
*fakeSite
|
|
path string
|
|
reached chan struct{}
|
|
release chan struct{}
|
|
once sync.Once
|
|
released sync.Once
|
|
}
|
|
|
|
func newBlockingSite(path string) *blockingSite {
|
|
return &blockingSite{fakeSite: newFakeSite(), path: path, reached: make(chan struct{}), release: make(chan struct{})}
|
|
}
|
|
|
|
func (s *blockingSite) RoundTrip(r *http.Request) (*http.Response, error) {
|
|
if r.URL.EscapedPath() == s.path {
|
|
s.once.Do(func() { close(s.reached) })
|
|
select {
|
|
case <-s.release:
|
|
case <-r.Context().Done():
|
|
return nil, r.Context().Err()
|
|
}
|
|
}
|
|
return s.fakeSite.RoundTrip(r)
|
|
}
|
|
|
|
func (s *blockingSite) unblock() { s.released.Do(func() { close(s.release) }) }
|
|
|
|
// lifecycleSite plants /, /a (no title: missing_title) and /b, listed in
|
|
// the sitemap, so the crawl fetches / and /a in its first batch and /b
|
|
// alone in the second.
|
|
func lifecycleSite(site *blockingSite) {
|
|
site.sitemap("/a", "/b")
|
|
site.plant(
|
|
doc{path: "/", title: "Home"},
|
|
doc{path: "/a", noTitle: true},
|
|
doc{path: "/b", title: "Page b"},
|
|
)
|
|
}
|
|
|
|
// crawlRow reads a crawl row.
|
|
func (e *testEnv) crawlRow(t *testing.T, id uint) models.Crawl {
|
|
t.Helper()
|
|
row, err := crawl.LoadCrawl(e.db, id)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return row
|
|
}
|
|
|
|
// runCrawlAsync runs the crawl job of id on ctx in a goroutine; the
|
|
// returned channel answers the job's error.
|
|
func (e *testEnv) runCrawlAsync(t *testing.T, ctx context.Context, id uint) <-chan error {
|
|
t.Helper()
|
|
job := e.crawlJob(t)
|
|
done := make(chan error, 1)
|
|
go func() { done <- job.Work(ctx, crawl.CrawlArgs{CrawlID: id}) }()
|
|
return done
|
|
}
|
|
|
|
// waitReached waits for the blocking request.
|
|
func waitReached(t *testing.T, site *blockingSite) {
|
|
t.Helper()
|
|
select {
|
|
case <-site.reached:
|
|
case <-time.After(20 * time.Second):
|
|
t.Fatal("the crawl never requested the blocking page")
|
|
}
|
|
}
|
|
|
|
// waitJob waits for the crawl job to return.
|
|
func waitJob(t *testing.T, done <-chan error) error {
|
|
t.Helper()
|
|
select {
|
|
case err := <-done:
|
|
return err
|
|
case <-time.After(30 * time.Second):
|
|
t.Fatal("the crawl job did not return")
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// storedPages lists the stored pages of a crawl with their status.
|
|
func (e *testEnv) storedPages(t *testing.T, id uint) map[string]int {
|
|
t.Helper()
|
|
var rows []struct {
|
|
URL string
|
|
Status int
|
|
}
|
|
if err := e.db.Raw(`SELECT url, status FROM golem15_seo_crawl_pages WHERE crawl_id = ?`, id).Scan(&rows).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
out := map[string]int{}
|
|
for _, r := range rows {
|
|
out[r.URL] = r.Status
|
|
}
|
|
return out
|
|
}
|
|
|
|
// crawlIssueKinds lists the issue kinds per address of a crawl.
|
|
func (e *testEnv) crawlIssueKinds(t *testing.T, id uint) map[string][]string {
|
|
t.Helper()
|
|
var rows []struct {
|
|
URL string
|
|
Kind string
|
|
}
|
|
if err := e.db.Raw(`SELECT url, kind FROM golem15_seo_issues WHERE crawl_id = ?`, id).Scan(&rows).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
out := map[string][]string{}
|
|
for _, r := range rows {
|
|
out[r.URL] = append(out[r.URL], r.Kind)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// TestCrawlCancelRunning pins L3 CR-01 and WR-01 (D-14): a running crawl
|
|
// that is cancelled stays running with cancel_requested until its job
|
|
// stops; meanwhile no other crawl starts, it cannot be deleted and Prune
|
|
// keeps it. When the job's context is cancelled (as River does), the job
|
|
// stores the pages fetched so far with their issues and ends cancelled
|
|
// with finished_at; the fetch cut short is not a network error.
|
|
func TestCrawlCancelRunning(t *testing.T) {
|
|
env := bootCrawlEnv(t)
|
|
token := env.admin(t, true)
|
|
if err := env.db.Exec(`UPDATE golem15_seo_settings SET crawl_delay = 100 WHERE id = 1`).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
site := newBlockingSite("/b")
|
|
defer site.unblock()
|
|
lifecycleSite(site)
|
|
id := env.startCrawl(t, token)
|
|
ctx, cancelJob := context.WithCancel(fetchguard.WithTransport(t.Context(), site))
|
|
defer cancelJob()
|
|
done := env.runCrawlAsync(t, ctx, id)
|
|
waitReached(t, site)
|
|
|
|
rec := env.expect(t, http.StatusOK, http.MethodPost, fmt.Sprintf("/_seo/api/v1/crawls/%d/cancel", id), token, map[string]any{})
|
|
var out struct {
|
|
Data map[string]any `json:"data"`
|
|
}
|
|
if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if out.Data["status"] != "running" || out.Data["cancel_requested"] != true {
|
|
t.Fatalf("cancel answer = %v, want running with cancel_requested", out.Data)
|
|
}
|
|
row := env.crawlRow(t, id)
|
|
if row.Status != models.CrawlRunning || row.CancelRequestedAt == nil || row.CancelledBy == "" || row.FinishedAt != nil {
|
|
t.Fatalf("row after the cancel = %+v", row)
|
|
}
|
|
if list := env.adminList(t, crawlsAPI, token); len(list.Data) == 0 || list.Data[0]["status"] != "Cancelling" {
|
|
t.Fatalf("the Crawls list shows %v, want Cancelling", list.Data)
|
|
}
|
|
if rec := env.call(t, http.MethodPost, "/_seo/api/v1/crawls", token, map[string]any{}); rec.Code != http.StatusConflict || !strings.Contains(rec.Body.String(), "crawls.already_running") {
|
|
t.Fatalf("a start while the cancelled crawl stops = %d %s", rec.Code, rec.Body.String())
|
|
}
|
|
if rec := env.call(t, http.MethodDelete, fmt.Sprintf("/_seo/api/v1/crawls?ids=%d", id), token, nil); rec.Code != http.StatusForbidden {
|
|
t.Fatalf("delete of the stopping crawl = %d %s", rec.Code, rec.Body.String())
|
|
}
|
|
if err := crawl.Prune(env.db, 1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if env.crawlRow(t, id).ID != id {
|
|
t.Fatal("Prune deleted the stopping crawl")
|
|
}
|
|
|
|
// River cancels the running job's context; the transport lets go.
|
|
cancelJob()
|
|
site.unblock()
|
|
if err := waitJob(t, done); err != nil {
|
|
t.Fatalf("the cancelled job answered %v", err)
|
|
}
|
|
row = env.crawlRow(t, id)
|
|
if row.Status != models.CrawlCancelled || row.FinishedAt == nil {
|
|
t.Fatalf("row after the job = %+v, want cancelled with finished_at", row)
|
|
}
|
|
pages := env.storedPages(t, id)
|
|
if pages["/"] != 200 || pages["/a"] != 200 {
|
|
t.Fatalf("stored pages = %v, want / and /a with 200", pages)
|
|
}
|
|
kinds := env.crawlIssueKinds(t, id)
|
|
if !strings.Contains(strings.Join(kinds["/a"], ","), "missing_title") {
|
|
t.Fatalf("issues of /a = %v, want missing_title", kinds["/a"])
|
|
}
|
|
for url, ks := range kinds {
|
|
for _, k := range ks {
|
|
if k == "page_status" || k == "sitemap_error" {
|
|
if url == "/b" {
|
|
t.Fatalf("the fetch cut short by the cancel is an issue: %s on %s", k, url)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
latest, err := crawl.LatestCrawl(env.db)
|
|
if err != nil || latest == nil || latest.ID != id {
|
|
t.Fatalf("LatestCrawl = %+v %v, want the cancelled crawl", latest, err)
|
|
}
|
|
}
|
|
|
|
// TestCrawlTimeout pins L3 CR-01: a crawl whose job context reaches its
|
|
// deadline stores what it fetched and ends failed with the time-limit
|
|
// error and finished_at; a later Start succeeds.
|
|
func TestCrawlTimeout(t *testing.T) {
|
|
env := bootCrawlEnv(t)
|
|
token := env.admin(t, true)
|
|
if err := env.db.Exec(`UPDATE golem15_seo_settings SET crawl_delay = 100 WHERE id = 1`).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
site := newBlockingSite("/b")
|
|
defer site.unblock()
|
|
lifecycleSite(site)
|
|
id := env.startCrawl(t, token)
|
|
ctx, cancel := context.WithTimeout(fetchguard.WithTransport(t.Context(), site), 3*time.Second)
|
|
defer cancel()
|
|
if err := waitJob(t, env.runCrawlAsync(t, ctx, id)); err != nil {
|
|
t.Fatalf("the timed-out job answered %v", err)
|
|
}
|
|
if !errors.Is(ctx.Err(), context.DeadlineExceeded) {
|
|
t.Fatal("the job returned before its deadline")
|
|
}
|
|
row := env.crawlRow(t, id)
|
|
if row.Status != models.CrawlFailed || !strings.Contains(row.Error, "time limit of 6h") || row.FinishedAt == nil {
|
|
t.Fatalf("row after the timeout = %+v", row)
|
|
}
|
|
if pages := env.storedPages(t, id); pages["/"] != 200 || pages["/a"] != 200 {
|
|
t.Fatalf("stored pages = %v, want / and /a", pages)
|
|
}
|
|
env.startCrawl(t, token)
|
|
}
|
|
|
|
// TestCrawlStoreTimeout pins 33-REVIEW-GAPFIX WR-03: a crawl whose
|
|
// result store runs past its bound (crawl.WithFinishTimeout here, the
|
|
// 10-minute FinishTimeout in production) ends at once, failed with the
|
|
// store error and finished_at, on a final write of its own; the job
|
|
// answers nil and a later Start succeeds.
|
|
func TestCrawlStoreTimeout(t *testing.T) {
|
|
env := bootCrawlEnv(t)
|
|
token := env.admin(t, true)
|
|
site := newFakeSite()
|
|
site.sitemap("/a")
|
|
site.plant(doc{path: "/", title: "Home"}, doc{path: "/a", title: "Page a"})
|
|
id := env.startCrawl(t, token)
|
|
|
|
// The store's first write (SavePages' insert of the pages) waits on
|
|
// this lock until the row has left running.
|
|
lock := env.db.Begin()
|
|
if lock.Error != nil {
|
|
t.Fatal(lock.Error)
|
|
}
|
|
var unlockOnce sync.Once
|
|
unlock := func() { unlockOnce.Do(func() { _ = lock.Rollback().Error }) }
|
|
t.Cleanup(unlock)
|
|
if err := lock.Exec(`LOCK TABLE golem15_seo_crawl_pages IN SHARE MODE`).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
ctx := crawl.WithFinishTimeout(fetchguard.WithTransport(t.Context(), site), 2*time.Second)
|
|
done := env.runCrawlAsync(t, ctx, id)
|
|
deadline := time.Now().Add(20 * time.Second)
|
|
row := env.crawlRow(t, id)
|
|
for row.Status == models.CrawlQueued || row.Status == models.CrawlRunning {
|
|
if time.Now().After(deadline) {
|
|
unlock()
|
|
t.Fatalf("the crawl is still %s 20 s after its store bound: %+v", row.Status, row)
|
|
}
|
|
time.Sleep(100 * time.Millisecond)
|
|
row = env.crawlRow(t, id)
|
|
}
|
|
unlock()
|
|
if row.Status != models.CrawlFailed || !strings.Contains(row.Error, "storing the crawl results took longer than 2s") ||
|
|
strings.Contains(row.Error, "time limit") || row.FinishedAt == nil {
|
|
t.Fatalf("row after the store timeout = %+v", row)
|
|
}
|
|
if err := waitJob(t, done); err != nil {
|
|
t.Fatalf("the job answered %v", err)
|
|
}
|
|
if pages := env.storedPages(t, id); len(pages) != 0 {
|
|
t.Fatalf("stored pages = %v, want none (the store rolled back)", pages)
|
|
}
|
|
env.startCrawl(t, token)
|
|
}
|
|
|
|
// TestCrawlStaleRecovery pins P33-60: a running crawl whose heartbeat is
|
|
// older than StaleAfter is ended by the next Start (failed with the
|
|
// no-heartbeat error, or cancelled when its cancel was requested), which
|
|
// then queues a new crawl; a fresh heartbeat still blocks Start.
|
|
func TestCrawlStaleRecovery(t *testing.T) {
|
|
env := bootCrawlEnv(t)
|
|
token := env.admin(t, true)
|
|
old := time.Now().UTC().Add(-11 * time.Minute)
|
|
insert := func(heartbeat time.Time, cancelRequested bool) uint {
|
|
t.Helper()
|
|
row := models.Crawl{Status: models.CrawlRunning, Trigger: models.TriggerAdmin, Site: siteURL, RenderStatus: models.RenderSkipped, StartedAt: &old, CreatedAt: &old, HeartbeatAt: &heartbeat}
|
|
if cancelRequested {
|
|
row.CancelRequestedAt = &heartbeat
|
|
row.CancelledBy = "Admin"
|
|
}
|
|
if err := env.db.Create(&row).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return row.ID
|
|
}
|
|
|
|
stale := insert(old, false)
|
|
next := env.startCrawl(t, token)
|
|
row := env.crawlRow(t, stale)
|
|
if row.Status != models.CrawlFailed || !strings.Contains(row.Error, "no heartbeat since") || row.FinishedAt == nil {
|
|
t.Fatalf("stale crawl after Start = %+v", row)
|
|
}
|
|
if env.crawlRow(t, next).Status != models.CrawlQueued {
|
|
t.Fatal("Start queued no new crawl")
|
|
}
|
|
if err := env.db.Exec(`UPDATE golem15_seo_crawls SET status = 'finished', finished_at = NOW() WHERE id = ?`, next).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
cancelled := insert(old, true)
|
|
env.startCrawl(t, token)
|
|
if row := env.crawlRow(t, cancelled); row.Status != models.CrawlCancelled || row.FinishedAt == nil || row.Error != "" {
|
|
t.Fatalf("stale crawl with a requested cancel = %+v", row)
|
|
}
|
|
if err := env.db.Exec(`UPDATE golem15_seo_crawls SET status = 'finished', finished_at = NOW() WHERE status = 'queued'`).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
fresh := insert(time.Now().UTC(), false)
|
|
if rec := env.call(t, http.MethodPost, "/_seo/api/v1/crawls", token, map[string]any{}); rec.Code != http.StatusConflict {
|
|
t.Fatalf("Start beside a crawl with a fresh heartbeat = %d %s", rec.Code, rec.Body.String())
|
|
}
|
|
if env.crawlRow(t, fresh).Status != models.CrawlRunning {
|
|
t.Fatal("a crawl with a fresh heartbeat was ended")
|
|
}
|
|
}
|