golem15/sm-seo-plugin
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: a crawl cancel waits for its job, and a stale crawl is ended

Browse Source 
Cancelling a running crawl records cancel_requested_at and leaves the
row running until its job stops and writes cancelled, so no second
crawl starts and the row cannot be deleted or pruned meanwhile. The job
writes heartbeat_at every minute; a running crawl without a heartbeat
for 10 minutes is ended by the next start or delete.
This commit is contained in:
Jakub Zych
2026-11-04 22:51:34 +01:00
parent d7597bb1a5
commit cd7182395f
12 changed files with 351 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
classes/crawl/actions.go
View File

@@ -71,11 +71,16 @@ func Unignore(tx *gorm.DB, issues []models.Issue) (int, error) {
}
// DeleteCrawls deletes crawls with their pages, links and issues; a
// queued or running one refuses the whole call with ErrDeleteRunning.
// queued or running one (also one whose cancel is requested) refuses the
// whole call with ErrDeleteRunning. A running crawl without a heartbeat is
// ended first (RecoverStale), so it can be deleted.
func DeleteCrawls(tx *gorm.DB, ids []uint) (int, error) {
if len(ids) == 0 {
return 0, nil
}
if _, err := RecoverStale(tx, time.Now()); err != nil {
return 0, err
}
var active int64
if err := tx.Model(&models.Crawl{}).Where("id IN ? AND status IN ?", ids, []string{models.CrawlQueued, models.CrawlRunning}).Count(&active).Error; err != nil {
return 0, err

124
classes/crawl/job.go
View File

@@ -45,12 +45,19 @@ func (CrawlArgs) Kind() string { return CrawlKind }
// The crawl job's time bounds (P33-60). JobTimeout is the job's deadline;
// FinishTimeout bounds storing what a crawl found, on a context the job's
// cancellation cannot reach.
// cancellation cannot reach; the job writes heartbeat_at every
// HeartbeatEvery, and a running crawl without a heartbeat for StaleAfter
// is ended by the next Start or delete (RecoverStale).
const (
JobTimeout = 6 * time.Hour
FinishTimeout = 10 * time.Minute
JobTimeout = 6 * time.Hour
FinishTimeout = 10 * time.Minute
HeartbeatEvery = time.Minute
StaleAfter = 10 * time.Minute
)
// heartbeatWrite bounds one heartbeat update.
const heartbeatWrite = 10 * time.Second
// The crawl limits used until the Site audit settings say otherwise.
const (
DefaultMaxPages = 5000
@@ -78,9 +85,11 @@ func (e *AlreadyRunningError) Since() time.Time {
// Start queues a crawl (D-14, P33-39) inside tx: the settings row is
// locked so two starts cannot both pass the check that no crawl is queued
// or running, then the crawl row is written and the conga job dispatched
// in the same transaction. A missing site address refuses with
// ErrNoSiteURL, a running crawl with *AlreadyRunningError.
// or running, a running crawl whose job stopped writing its heartbeat is
// ended (RecoverStale), then the crawl row is written and the conga job
// dispatched in the same transaction. A missing site address refuses with
// ErrNoSiteURL, a queued or running crawl (also one whose cancel is
// requested but whose job has not stopped) with *AlreadyRunningError.
func Start(ctx context.Context, svc *classes.Service, tx *gorm.DB, trigger, name string) (models.Crawl, error) {
site := svc.SiteURL()
if site == "" {
@@ -92,6 +101,11 @@ func Start(ctx context.Context, svc *classes.Service, tx *gorm.DB, trigger, name
if err := tx.Exec(`SELECT id FROM golem15_seo_settings WHERE id = 1 FOR UPDATE`).Error; err != nil {
return models.Crawl{}, err
}
stale, err := RecoverStale(tx, time.Now().UTC())
if err != nil {
return models.Crawl{}, err
}
endStaleJobs(ctx, svc, tx, stale)
active, err := ActiveCrawl(tx)
if err != nil {
return models.Crawl{}, err
@@ -182,6 +196,7 @@ func LoadCrawl(db *gorm.DB, id uint) (models.Crawl, error) {
// context) or at the job timeout; what was fetched is then stored on a
// context detached from it with its own FinishTimeout bound, so a
// cancelled or timed-out crawl keeps its results and never stays running.
// While it runs, the job writes heartbeat_at every HeartbeatEvery.
func Run(ctx context.Context, svc *classes.Service, id uint) error {
gdb, err := svc.GormDB()
if err != nil {
@@ -197,19 +212,49 @@ func Run(ctx context.Context, svc *classes.Service, id uint) error {
}
now := time.Now().UTC()
moved := db.Model(&models.Crawl{}).Where("id = ? AND status = ?", id, models.CrawlQueued).
Updates(map[string]any{"status": models.CrawlRunning, "started_at": now})
Updates(map[string]any{"status": models.CrawlRunning, "started_at": now, "heartbeat_at": now})
if moved.Error != nil {
return moved.Error
}
if moved.RowsAffected == 0 {
return nil
}
stop := heartbeat(ctx, gdb, id)
defer stop()
res, runErr := run(ctx, svc, db, row)
fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), FinishTimeout)
defer cancel()
return finish(fctx, svc, gdb.WithContext(fctx), row, res, runErr, ctx.Err())
}
// heartbeat writes the crawl's heartbeat_at every HeartbeatEvery until the
// returned stop is called; each write runs on a context detached from the
// job's, bounded by heartbeatWrite.
func heartbeat(ctx context.Context, gdb *gorm.DB, id uint) (stop func()) {
done := make(chan struct{})
finished := make(chan struct{})
go func() {
defer close(finished)
t := time.NewTicker(HeartbeatEvery)
defer t.Stop()
for {
select {
case <-done:
return
case <-t.C:
hctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), heartbeatWrite)
_ = gdb.WithContext(hctx).Model(&models.Crawl{}).Where("id = ? AND status = ?", id, models.CrawlRunning).
Update("heartbeat_at", time.Now().UTC()).Error
cancel()
}
}
}()
return func() {
close(done)
<-finished
}
}
// run fetches and checks the site of one crawl.
func run(ctx context.Context, svc *classes.Service, db *gorm.DB, row models.Crawl) (*Result, error) {
f, err := NewFetcher(row.Site, svc.Config())
@@ -267,14 +312,17 @@ func run(ctx context.Context, svc *classes.Service, db *gorm.DB, row models.Craw
return Crawl(ctx, f, opt)
}
// cancelRequested reports whether the crawl was cancelled: its status is
// cancelled.
// cancelRequested reports whether a cancel of the crawl was requested: its
// cancel_requested_at is set, or its status is cancelled.
func cancelRequested(db *gorm.DB, id uint) bool {
var status string
if err := db.Raw(`SELECT status FROM golem15_seo_crawls WHERE id = ?`, id).Scan(&status).Error; err != nil {
var rows []struct {
Status string
Requested bool
}
if err := db.Raw(`SELECT status, cancel_requested_at IS NOT NULL AS requested FROM golem15_seo_crawls WHERE id = ?`, id).Scan(&rows).Error; err != nil || len(rows) == 0 {
return false
}
return status == models.CrawlCancelled
return rows[0].Requested || rows[0].Status == models.CrawlCancelled
}
// timeLimitError is the error of a crawl that reached JobTimeout.
@@ -370,6 +418,31 @@ func finish(ctx context.Context, svc *classes.Service, db *gorm.DB, row models.C
return nil
}
// endStaleJobs fails or stops the conga jobs of crawls RecoverStale ended,
// once tx commits; best effort.
func endStaleJobs(ctx context.Context, svc *classes.Service, tx *gorm.DB, rows []models.Crawl) {
if len(rows) == 0 {
return
}
lagoon.AfterCommit(ctx, tx, func(ctx context.Context, _ *gorm.DB) {
m, err := conga.From(svc.App())
if err != nil {
return
}
for _, r := range rows {
if r.JobID == nil {
continue
}
meta := map[string]any{"crawl_id": r.ID, "status": r.Status}
if r.Status == models.CrawlCancelled {
_ = m.StopJob(ctx, *r.JobID, meta)
} else {
_ = m.FailJob(ctx, *r.JobID, meta)
}
}
})
}
// Lookup returns the CMS record behind a crawled address (cached per
// address): the content type whose Match answers the path exactly, its
// descriptor facts in the matched locale, its primary parent's address and
@@ -455,20 +528,31 @@ func Keyphrases(db *gorm.DB) (map[string]string, error) {
// running.
var errCancelledCrawl = errors.New("golem15.seo: the crawl is not running")
// Cancel stops a queued or running crawl inside tx: the row is marked
// cancelled by name (the crawl keeps what it fetched) and its conga job
// cancelled.
// Cancel stops a queued or running crawl inside tx. A queued crawl ends
// cancelled at once (by name, finished_at set). A running crawl gets
// cancel_requested_at and cancelled_by and stays running: its job stops
// between requests and writes cancelled with the pages and issues fetched
// so far, and until then no other crawl starts. The conga job is cancelled
// in both cases.
func Cancel(ctx context.Context, svc *classes.Service, tx *gorm.DB, row models.Crawl, name string) error {
if !row.Active() {
return errCancelledCrawl
}
now := time.Now().UTC()
updates := map[string]any{"status": models.CrawlCancelled, "cancelled_by": name}
if row.Status == models.CrawlQueued {
updates["finished_at"] = now
queued := tx.Model(&models.Crawl{}).Where("id = ? AND status = ?", row.ID, models.CrawlQueued).
Updates(map[string]any{"status": models.CrawlCancelled, "cancelled_by": name, "finished_at": now})
if queued.Error != nil {
return queued.Error
}
if err := tx.Model(&models.Crawl{}).Where("id = ?", row.ID).Updates(updates).Error; err != nil {
return err
if queued.RowsAffected == 0 {
running := tx.Model(&models.Crawl{}).Where("id = ? AND status = ?", row.ID, models.CrawlRunning).
Updates(map[string]any{"cancel_requested_at": now, "cancelled_by": name})
if running.Error != nil {
return running.Error
}
if running.RowsAffected == 0 {
return errCancelledCrawl
}
}
if row.JobID != nil {
if m, err := conga.From(svc.App()); err == nil {

49
classes/crawl/store.go
View File

@@ -189,6 +189,55 @@ func ActiveCrawl(db *gorm.DB) (*models.Crawl, error) {
return &rows[0], nil
}
// RecoverStale ends the running crawls whose job stopped writing its
// heartbeat (P33-60): a running row whose COALESCE(heartbeat_at,
// started_at, created_at) is older than StaleAfter before now becomes
// cancelled when its cancel was requested, else failed with the error
// "the crawl job stopped without finishing (no heartbeat since <time>)",
// with finished_at now. It answers the rows it ended, with their new
// status, so the caller can end their conga jobs.
func RecoverStale(tx *gorm.DB, now time.Time) ([]models.Crawl, error) {
now = now.UTC()
var rows []models.Crawl
if err := tx.Where("status = ? AND COALESCE(heartbeat_at, started_at, created_at) < ?", models.CrawlRunning, now.Add(-StaleAfter)).
Order("id ASC").Find(&rows).Error; err != nil {
return nil, err
}
out := rows[:0]
for _, r := range rows {
last := r.HeartbeatAt
if last == nil {
last = r.StartedAt
}
if last == nil {
last = r.CreatedAt
}
updates := map[string]any{"finished_at": now}
if r.CancelRequestedAt != nil {
updates["status"], updates["error"] = models.CrawlCancelled, ""
} else {
since := "an unknown time"
if last != nil {
since = last.UTC().Format(time.RFC3339)
}
updates["status"], updates["error"] = models.CrawlFailed, "the crawl job stopped without finishing (no heartbeat since "+since+")"
}
res := tx.Model(&models.Crawl{}).Where("id = ? AND status = ?", r.ID, models.CrawlRunning).Updates(updates)
if res.Error != nil {
return nil, res.Error
}
if res.RowsAffected == 0 {
continue
}
r.Status = updates["status"].(string)
r.Error = updates["error"].(string)
r.FinishedAt = &now
pluginLog.Warn("golem15.seo: ended a stale crawl", "crawl", r.ID, "status", r.Status)
out = append(out, r)
}
return out, nil
}
// CurrentScope narrows an issues query to the audit's current issues: the
// newest finished, failed or cancelled crawl's, or before any crawl the
// stored CMS-data issues.