orchard

  1package main
  2
  3import (
  4	"context"
  5	"database/sql"
  6	"encoding/json"
  7	"fmt"
  8	"log/slog"
  9	"time"
 10)
 11
 12const (
 13	cycleInterval = 30 * time.Second
 14
 15	// Two pools, so a nine-minute crawl cannot starve a three-minute ping.
 16	fastWorkers = 2
 17	slowWorkers = 2
 18
 19	checkInterval      = 3 * time.Minute
 20	lighthouseInterval = 24 * time.Hour
 21	crawlInterval      = 7 * 24 * time.Hour
 22
 23	// Watchdog cutoffs, both past the work's own deadline, so a slow but living
 24	// job records its own result instead of being declared dead underneath it.
 25	crawlWedgeAfter      = 15 * time.Minute
 26	lighthouseWedgeAfter = 5 * time.Minute
 27
 28	cleanupInterval = 24 * time.Hour
 29	checkRetention  = 3 * 24 * time.Hour
 30)
 31
 32// Scheduler owns the background work.
 33type Scheduler struct {
 34	db       *sql.DB
 35	notifier *Notifier
 36	root     string
 37
 38	// Buffered channels as semaphores: a full channel blocks the goroutine that
 39	// wanted to work rather than spawning an unbounded number of them.
 40	fast chan struct{}
 41	slow chan struct{}
 42}
 43
 44func NewScheduler(db *sql.DB, notifier *Notifier, root string) *Scheduler {
 45	return &Scheduler{
 46		db:       db,
 47		notifier: notifier,
 48		root:     root,
 49		fast:     make(chan struct{}, fastWorkers),
 50		slow:     make(chan struct{}, slowWorkers),
 51	}
 52}
 53
 54// ResetOnBoot clears rows left queued or running by a previous process, whose
 55// goroutines are gone; without it a property interrupted by a deploy sticks.
 56func (s *Scheduler) ResetOnBoot(ctx context.Context) error {
 57	if _, err := s.db.ExecContext(ctx,
 58		"UPDATE properties SET crawl_state = 'idle' WHERE crawl_state IN ('queued', 'running')"); err != nil {
 59		return err
 60	}
 61	_, err := s.db.ExecContext(ctx,
 62		"UPDATE properties SET lighthouse_state = 'idle' WHERE lighthouse_state IN ('queued', 'running')")
 63	return err
 64}
 65
 66// Run loops until the context is cancelled.
 67func (s *Scheduler) Run(ctx context.Context) {
 68	ticker := time.NewTicker(cycleInterval)
 69	defer ticker.Stop()
 70
 71	var lastCleanup time.Time
 72
 73	for {
 74		s.cycle(ctx, &lastCleanup)
 75
 76		select {
 77		case <-ctx.Done():
 78			slog.Info("stopping", slog.String("component", "scheduler"))
 79			return
 80		case <-ticker.C:
 81		}
 82	}
 83}
 84
 85func (s *Scheduler) cycle(ctx context.Context, lastCleanup *time.Time) {
 86	// Each step logs and continues rather than returning, so a failure
 87	// enqueuing Lighthouse does not also skip the uptime checks.
 88	if err := s.enqueueChecks(ctx); err != nil {
 89		slog.Info(fmt.Sprintf("enqueue checks: %v", err), slog.String("component", "scheduler"))
 90	}
 91	if err := s.enqueueLighthouse(ctx); err != nil {
 92		slog.Info(fmt.Sprintf("enqueue lighthouse: %v", err), slog.String("component", "scheduler"))
 93	}
 94	if err := s.enqueueCrawls(ctx); err != nil {
 95		slog.Info(fmt.Sprintf("enqueue crawls: %v", err), slog.String("component", "scheduler"))
 96	}
 97	if err := s.resetWedged(ctx); err != nil {
 98		slog.Info(fmt.Sprintf("reset wedged: %v", err), slog.String("component", "scheduler"))
 99	}
100	if err := s.maybeCleanup(ctx, lastCleanup); err != nil {
101		slog.Info(fmt.Sprintf("cleanup: %v", err), slog.String("component", "scheduler"))
102	}
103}
104
105// due returns the properties matching a condition, all columns loaded.
106func (s *Scheduler) due(ctx context.Context, where string, args ...any) ([]*Property, error) {
107	rows, err := s.db.QueryContext(ctx,
108		"SELECT "+propertyColumns+" FROM properties WHERE "+where, args...)
109	if err != nil {
110		return nil, err
111	}
112	defer rows.Close()
113
114	var out []*Property
115	for rows.Next() {
116		p, err := scanProperty(rows.Scan)
117		if err != nil {
118			return nil, err
119		}
120		out = append(out, p)
121	}
122	return out, rows.Err()
123}
124
125// enqueueChecks starts the HTTP probes that are due. The next run time is
126// written before the work starts, so the next tick does not probe it again.
127func (s *Scheduler) enqueueChecks(ctx context.Context) error {
128	now := nowMS()
129	props, err := s.due(ctx,
130		"last_run_at IS NULL OR next_run_at IS NULL OR next_run_at <= ?", now)
131	if err != nil {
132		return err
133	}
134
135	for _, p := range props {
136		if _, err := s.db.ExecContext(ctx,
137			"UPDATE properties SET next_run_at = ?, last_run_at = ?, updated_at = ? WHERE id = ?",
138			next3MinBoundary(), now, now, p.ID[:]); err != nil {
139			slog.Info(fmt.Sprintf("claim check for %s: %v", p.URL, err), slog.String("component", "scheduler"))
140			continue
141		}
142
143		go func(p *Property) {
144			select {
145			case s.fast <- struct{}{}:
146				defer func() { <-s.fast }()
147			case <-ctx.Done():
148				return
149			}
150			slog.Info(fmt.Sprintf("checking %s", p.URL), slog.String("component", "scheduler"))
151			if err := processCheck(ctx, s.db, s.notifier, p); err != nil {
152				slog.Error(fmt.Sprintf("check failed for %s: %v", p.URL, err), slog.String("component", "scheduler"))
153			}
154		}(p)
155	}
156	return nil
157}
158
159func (s *Scheduler) enqueueLighthouse(ctx context.Context) error {
160	now := nowMS()
161	props, err := s.due(ctx,
162		`(last_lighthouse_run_at IS NULL OR next_lighthouse_run_at IS NULL
163		  OR next_lighthouse_run_at <= ?)
164		 AND lighthouse_state NOT IN ('queued', 'running')`, now)
165	if err != nil {
166		return err
167	}
168
169	for _, p := range props {
170		next := now + lighthouseInterval.Milliseconds()
171		if _, err := s.db.ExecContext(ctx,
172			`UPDATE properties SET next_lighthouse_run_at = ?, last_lighthouse_run_at = ?,
173			  lighthouse_state = 'queued', updated_at = ? WHERE id = ?`,
174			next, now, now, p.ID[:]); err != nil {
175			slog.Info(fmt.Sprintf("claim lighthouse for %s: %v", p.URL, err), slog.String("component", "scheduler"))
176			continue
177		}
178
179		go func(p *Property) {
180			select {
181			case s.slow <- struct{}{}:
182				defer func() { <-s.slow }()
183			case <-ctx.Done():
184				return
185			}
186			s.runLighthouseFor(ctx, p)
187		}(p)
188	}
189	return nil
190}
191
192func (s *Scheduler) runLighthouseFor(ctx context.Context, p *Property) {
193	slog.Info(fmt.Sprintf("lighthouse %s", p.URL), slog.String("component", "scheduler"))
194	started := time.Now()
195
196	now := nowMS()
197	if _, err := s.db.ExecContext(ctx,
198		`UPDATE properties SET lighthouse_state = 'running', lighthouse_started_at = ?,
199		  updated_at = ? WHERE id = ?`, now, now, p.ID[:]); err != nil {
200		slog.Info(fmt.Sprintf("mark lighthouse running for %s: %v", p.URL, err), slog.String("component", "scheduler"))
201	}
202
203	fail := func(err error) {
204		slog.Error(fmt.Sprintf("lighthouse failed for %s: %v", p.URL, err), slog.String("component", "scheduler"))
205		if _, dbErr := s.db.ExecContext(ctx,
206			`UPDATE properties SET lighthouse_state = 'idle', last_lighthouse_error = ?,
207			  last_lighthouse_duration_ms = ?, updated_at = ? WHERE id = ?`,
208			err.Error(), elapsedMS(started), nowMS(), p.ID[:]); dbErr != nil {
209			slog.Error(fmt.Sprintf("record lighthouse error for %s: %v", p.URL, dbErr), slog.String("component", "scheduler"))
210		}
211	}
212
213	report, err := runLighthouse(ctx, s.root, p.URL)
214	if err != nil {
215		fail(err)
216		return
217	}
218	scores, err := parseScores(report)
219	if err != nil {
220		fail(err)
221		return
222	}
223
224	scoresJSON, err := json.Marshal(scores)
225	if err != nil {
226		fail(err)
227		return
228	}
229	// Details are best effort: a Lighthouse release that reshapes auditRefs must
230	// not throw away a good audit. The template tests for "null".
231	detailsJSON := []byte("null")
232	if details := parseDetails(report); details != nil {
233		if encoded, err := json.Marshal(details); err == nil {
234			detailsJSON = encoded
235		}
236	}
237
238	if _, err := s.db.ExecContext(ctx,
239		`UPDATE properties SET lighthouse_scores = ?, lighthouse_details = ?,
240		  last_lighthouse_success_at = ?, last_lighthouse_error = NULL,
241		  last_lighthouse_duration_ms = ?, lighthouse_state = 'idle', updated_at = ?
242		 WHERE id = ?`,
243		string(scoresJSON), string(detailsJSON), nowMS(), elapsedMS(started), nowMS(), p.ID[:]); err != nil {
244		slog.Info(fmt.Sprintf("store lighthouse for %s: %v", p.URL, err), slog.String("component", "scheduler"))
245	}
246}
247
248func (s *Scheduler) enqueueCrawls(ctx context.Context) error {
249	now := nowMS()
250	props, err := s.due(ctx,
251		`(last_run_at_crawler IS NULL OR next_run_at_crawler IS NULL
252		  OR next_run_at_crawler <= ?)
253		 AND crawl_state NOT IN ('queued', 'running')`, now)
254	if err != nil {
255		return err
256	}
257
258	for _, p := range props {
259		next := now + crawlInterval.Milliseconds()
260		if _, err := s.db.ExecContext(ctx,
261			`UPDATE properties SET next_run_at_crawler = ?, last_run_at_crawler = ?,
262			  crawl_state = 'queued', updated_at = ? WHERE id = ?`,
263			next, now, now, p.ID[:]); err != nil {
264			slog.Info(fmt.Sprintf("claim crawl for %s: %v", p.URL, err), slog.String("component", "scheduler"))
265			continue
266		}
267
268		go func(p *Property) {
269			select {
270			case s.slow <- struct{}{}:
271				defer func() { <-s.slow }()
272			case <-ctx.Done():
273				return
274			}
275			s.runCrawlFor(ctx, p)
276		}(p)
277	}
278	return nil
279}
280
281func (s *Scheduler) runCrawlFor(ctx context.Context, p *Property) {
282	slog.Info(fmt.Sprintf("crawling %s", p.URL), slog.String("component", "scheduler"))
283	started := time.Now()
284
285	now := nowMS()
286	if _, err := s.db.ExecContext(ctx,
287		`UPDATE properties SET crawl_state = 'running', crawl_started_at = ?,
288		  last_crawl_pages_count = 0, updated_at = ? WHERE id = ?`,
289		now, now, p.ID[:]); err != nil {
290		slog.Info(fmt.Sprintf("mark crawl running for %s: %v", p.URL, err), slog.String("component", "scheduler"))
291	}
292
293	// Synchronous rather than a goroutine per update, which would leave detached
294	// writers contending for SQLite's single write lock on one column.
295	progress := func(pages int) {
296		if _, err := s.db.ExecContext(ctx,
297			"UPDATE properties SET last_crawl_pages_count = ? WHERE id = ?",
298			int64(pages), p.ID[:]); err != nil {
299			slog.Info(fmt.Sprintf("crawl progress for %s: %v", p.URL, err), slog.String("component", "scheduler"))
300		}
301	}
302
303	insights, err := RunSEOSpider(ctx, p.URL, progress)
304	if err != nil {
305		slog.Error(fmt.Sprintf("crawl failed for %s: %v", p.URL, err), slog.String("component", "scheduler"))
306		if _, dbErr := s.db.ExecContext(ctx,
307			`UPDATE properties SET crawl_state = 'idle', last_crawl_error = ?,
308			  last_crawl_duration_ms = ?, updated_at = ? WHERE id = ?`,
309			err.Error(), elapsedMS(started), nowMS(), p.ID[:]); dbErr != nil {
310			slog.Error(fmt.Sprintf("record crawl error for %s: %v", p.URL, dbErr), slog.String("component", "scheduler"))
311		}
312		return
313	}
314
315	encoded, err := json.Marshal(insights)
316	if err != nil {
317		slog.Info(fmt.Sprintf("encode insights for %s: %v", p.URL, err), slog.String("component", "scheduler"))
318		encoded = []byte("[]")
319	}
320
321	if _, err := s.db.ExecContext(ctx,
322		`UPDATE properties SET crawler_insights = ?, crawl_state = 'idle',
323		  last_crawl_success_at = ?, last_crawl_error = NULL,
324		  last_crawl_duration_ms = ?, updated_at = ? WHERE id = ?`,
325		string(encoded), nowMS(), elapsedMS(started), nowMS(), p.ID[:]); err != nil {
326		slog.Info(fmt.Sprintf("store insights for %s: %v", p.URL, err), slog.String("component", "scheduler"))
327	}
328}
329
330// resetWedged is the watchdog; the thresholds are the cutoff constants above.
331func (s *Scheduler) resetWedged(ctx context.Context) error {
332	now := nowMS()
333
334	if _, err := s.db.ExecContext(ctx,
335		`UPDATE properties SET crawl_state = 'idle',
336		   last_crawl_error = 'Crawl timed out or was interrupted'
337		 WHERE crawl_state = 'running' AND crawl_started_at IS NOT NULL
338		   AND crawl_started_at < ?`,
339		now-crawlWedgeAfter.Milliseconds()); err != nil {
340		return err
341	}
342
343	_, err := s.db.ExecContext(ctx,
344		`UPDATE properties SET lighthouse_state = 'idle',
345		   last_lighthouse_error = 'Lighthouse run timed out or was interrupted'
346		 WHERE lighthouse_state = 'running' AND lighthouse_started_at IS NOT NULL
347		   AND lighthouse_started_at < ?`,
348		now-lighthouseWedgeAfter.Milliseconds())
349	return err
350}
351
352// maybeCleanup deletes checks past the retention window, once a day. The timer
353// is in memory, so it deletes everything past the cutoff and catches up.
354func (s *Scheduler) maybeCleanup(ctx context.Context, last *time.Time) error {
355	if !last.IsZero() && time.Since(*last) < cleanupInterval {
356		return nil
357	}
358
359	result, err := s.db.ExecContext(ctx,
360		"DELETE FROM checks WHERE created_at < ?", nowMS()-checkRetention.Milliseconds())
361	if err != nil {
362		return err
363	}
364	*last = time.Now()
365
366	if n, err := result.RowsAffected(); err == nil && n > 0 {
367		slog.Info(fmt.Sprintf("deleted %d checks older than %s", n, checkRetention), slog.String("component", "scheduler"))
368	}
369	return nil
370}
371
372// elapsedMS is in whole milliseconds, the unit every *_duration_ms column uses.
373func elapsedMS(started time.Time) int64 {
374	return time.Since(started).Milliseconds()
375}