feat(surf): serve startup summary, shutdown report and http request log
- serve prints a startup summary on a terminal (version, environment, URL, admin, database host and name, pending migrations, plugins, drivers) and logs it as a 'startup summary' record plus 'listening on <addr>' - on SIGINT/SIGTERM serve drains, prints and logs a shutdown report and returns nil (exit 0), or the drain errors (exit 1) - the request log is on by default under diary component http with request_id; the route pattern, never the path, is logged - lagoon.PendingMigrations, conga.Worker.Stats, diary.HideOnTTY, bonfire.IsTerminal and bonfire.Details back the summary and report
This commit is contained in:
@@ -34,7 +34,7 @@ Use cases:
|
||||
- The WinterCMS job manager operations with the same semantics: `conga.Manager.StartJob`, `conga.Manager.UpdateJobState`, `conga.Manager.UpdateMetadata`, `conga.Manager.CompleteJob`, `conga.Manager.FailJob`, `conga.Manager.CheckIfCanceled` and `conga.Manager.GetMetadata`. Updates are raw column writes, so `updated_at` changes only on dispatch and `conga.Manager.StartJob`.
|
||||
- Cancellation in two parts: `conga.Manager.CancelJob` is the outside cancel (sets `is_canceled` and `conga.StatusStopped`, then cancels the River job, so a queued job never starts and a running job's context is cancelled); `conga.Manager.StopJob` is what a job calls on its own row after `conga.Manager.CheckIfCanceled` reports true (status only, the WinterCMS `cancelJob`).
|
||||
- Outcome rules in the worker: an error on an attempt before the last leaves the row in progress so River can retry; the final failed attempt, or a recovered panic on it, sets `conga.StatusError` with the error text under the metadata key `error`; an error on a row that was stopped or cancelled cancels the River job instead. A job that returns nil without completing its row leaves it as it is. Skipped work is recorded as complete with `{"skipped": true}` metadata.
|
||||
- Workers: `conga.StartWorker` registers every plugin job and starts one River client; `conga.WorkerOptions.Queues` limits it to some queues, and an unknown queue is `conga.ErrUnknownQueue` listing the known ones. `conga.Worker.Stop` stops it gracefully and cancels running jobs when its context ends. `conga.StartServeWorker` is the variant the `serve` command uses: it starts nothing when `queue.work_in_serve` is false. An app without jobs still gets a worker that starts and idles.
|
||||
- Workers: `conga.StartWorker` registers every plugin job and starts one River client; `conga.WorkerOptions.Queues` limits it to some queues, and an unknown queue is `conga.ErrUnknownQueue` listing the known ones. `conga.Worker.Stop` stops it gracefully and cancels running jobs when its context ends; `conga.Worker.Stats` counts attempts done, failed, in flight and handed back to the queue by that cancellation. `conga.StartServeWorker` is the variant the `serve` command uses: it starts nothing when `queue.work_in_serve` is false. An app without jobs still gets a worker that starts and idles.
|
||||
- Scheduled commands: every worker carries one River periodic job per `pact.HasSchedule` entry, in plugin activation order then declaration order, with the id `<plugin id>[<index>]:<command>`. Only the elected leader enqueues. Each run is a `conga.ScheduledCommandArgs` job on `conga.QueueScheduled` with one attempt (an interrupted run is not retried; the next period runs normally) and unique by args within its cadence period, so a leader failover cannot double-enqueue a period. The worker runs a job only when its entry exists in the compiled schedule and its command and arguments match that entry exactly, so a forged `river_job` row cannot run an arbitrary command. An unregistered command, or an app with no published `bonfire.Catalog`, is logged at Warn (`schedule: command not registered; skipping`) and skipped without failing the worker or other entries. Command output is logged line by line at Info with a `command` attribute; failures are logged with the duration.
|
||||
- Framework schedule entries: before the plugin entries, every worker carries the framework's own entries from `lagoon.FrameworkSchedule`, listed under the plugin id `lagoon.FrameworkScheduleID`. Today that is the daily `deferred:purge` at `database.deferred_bindings.purge_at` (default `03:00`), with the id `summercms.lagoon[0]:deferred:purge`; it runs, appears in `schedule:run --once` and is protected by the compiled entry table exactly like a plugin entry. An empty `purge_at` removes it, and a malformed one fails the worker start.
|
||||
- Wall-clock schedules: `conga.Daily` fires at the next `hh:mm` in its location and `conga.Every` at the next multiple of its interval since local midnight, so a restart never delays a daily run by up to a day the way `river.PeriodicInterval(24h)` would. On a DST day `conga.Daily` keeps the wall-clock time. A schedule entry with an empty command, a zero cadence, a daily time out of range, or an interval under one second or not dividing 24h fails the worker start with an error naming the plugin id and entry index.
|
||||
@@ -183,7 +183,8 @@ defer w.Stop(context.Background())
|
||||
| `conga.StartServeWorker` | The worker of the `serve` command; nil when `queue.work_in_serve` is false. |
|
||||
| `conga.WorkerOptions` | Selects the queues a worker runs. |
|
||||
| `conga.RuntimeCommands` | Returns the `queue:work`, `schedule:run` and `queue:clear` commands. |
|
||||
| `conga.Worker` | A running worker; `conga.Worker.Stop` stops it and `conga.Worker.Queues` lists its queues. |
|
||||
| `conga.Worker` | A running worker; `conga.Worker.Stop` stops it, `conga.Worker.Queues` lists its queues and `conga.Worker.Stats` returns its attempt counters. |
|
||||
| `conga.WorkerStats` | Attempt counters since the worker started: `Done`, `Failed`, `InFlight` and `HandedBack` (attempts cancelled when `Stop` reached its deadline; River records them as errored attempts and retries them). The shutdown reports of `serve` and `queue:work` print them. |
|
||||
| `conga.Daily` | `river.PeriodicSchedule` firing at `Hour:Minute` every day in `Loc` (UTC when nil). |
|
||||
| `conga.Every` | `river.PeriodicSchedule` firing at every multiple of `Interval` since local midnight in `Loc` (UTC when nil). |
|
||||
| `conga.ScheduledCommandArgs` | Args of one scheduled run: compiled `Entry` id, `Command` and `Args`; kind `summer.scheduled_command`. |
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/backpack"
|
||||
@@ -55,6 +56,9 @@ type Manager struct {
|
||||
// compiled entry table of the running worker, keyed by entry id.
|
||||
scheduledJob pact.Job
|
||||
schedule map[string]pact.ScheduledCommand
|
||||
|
||||
// counters are the attempt counters of the running worker.
|
||||
counters atomic.Pointer[workerCounters]
|
||||
}
|
||||
|
||||
// From returns the app's Manager, publishing a new one on first use.
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"runtime/debug"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/backpack"
|
||||
@@ -38,6 +39,44 @@ type Worker struct {
|
||||
client *river.Client[*sql.Tx]
|
||||
listener *pgxpool.Pool
|
||||
queues []string
|
||||
counters *workerCounters
|
||||
}
|
||||
|
||||
// WorkerStats counts the job attempts a worker ran since it started.
|
||||
type WorkerStats struct {
|
||||
// Done is the number of attempts that returned no error.
|
||||
Done int64
|
||||
// Failed is the number of attempts that returned an error, other than
|
||||
// those cancelled by Stop (HandedBack).
|
||||
Failed int64
|
||||
// InFlight is the number of attempts running now.
|
||||
InFlight int64
|
||||
// HandedBack is the number of attempts cancelled by Stop after its
|
||||
// deadline (River's StopAndCancel). River records the cancellation as
|
||||
// an errored attempt, so the job is retried later, or fails for good
|
||||
// when it was its last attempt.
|
||||
HandedBack int64
|
||||
}
|
||||
|
||||
// workerCounters are the live counters behind WorkerStats.
|
||||
type workerCounters struct {
|
||||
done, failed, inFlight, handedBack atomic.Int64
|
||||
// stopping is set by Stop before running jobs are cancelled.
|
||||
stopping atomic.Bool
|
||||
}
|
||||
|
||||
// Stats returns the worker's attempt counters. A nil worker has none.
|
||||
func (w *Worker) Stats() WorkerStats {
|
||||
if w == nil || w.counters == nil {
|
||||
return WorkerStats{}
|
||||
}
|
||||
c := w.counters
|
||||
return WorkerStats{
|
||||
Done: c.done.Load(),
|
||||
Failed: c.failed.Load(),
|
||||
InFlight: c.inFlight.Load(),
|
||||
HandedBack: c.handedBack.Load(),
|
||||
}
|
||||
}
|
||||
|
||||
// Queues returns the sorted queue names the worker runs.
|
||||
@@ -152,9 +191,11 @@ func StartWorker(ctx context.Context, app *backpack.App, plugins []party.Plugin,
|
||||
closePool(listener)
|
||||
return nil, fmt.Errorf("conga: start worker: %w", err)
|
||||
}
|
||||
counters := &workerCounters{}
|
||||
m.worker = client
|
||||
m.schedule = table
|
||||
return &Worker{m: m, client: client, listener: listener, queues: sortedQueueNames(queues)}, nil
|
||||
m.counters.Store(counters)
|
||||
return &Worker{m: m, client: client, listener: listener, queues: sortedQueueNames(queues), counters: counters}, nil
|
||||
}
|
||||
|
||||
// StartServeWorker starts the in-process worker of the serve command on
|
||||
@@ -175,6 +216,9 @@ func (w *Worker) Stop(ctx context.Context) error {
|
||||
}
|
||||
err := w.client.Stop(ctx)
|
||||
if err != nil && ctx.Err() != nil {
|
||||
if w.counters != nil {
|
||||
w.counters.stopping.Store(true)
|
||||
}
|
||||
hardCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
||||
err = w.client.StopAndCancel(hardCtx)
|
||||
cancel()
|
||||
@@ -235,8 +279,8 @@ func selectQueues(known map[string]int, want []string) (map[string]int, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// runAttempt runs one River attempt of a job and applies the summer_jobs
|
||||
// rules: the row id goes into ctx; a panic becomes an error; an error on a
|
||||
// runAttempt runs one River attempt of a job, counts it in the running
|
||||
// worker's WorkerStats and applies the summer_jobs rules: the row id goes into ctx; a panic becomes an error; an error on a
|
||||
// STOPPED or canceled row cancels the River job; an error on the final
|
||||
// attempt sets StatusError with the error text under metadata "error". An
|
||||
// error on an earlier attempt leaves the row as it is so River can retry.
|
||||
@@ -245,7 +289,22 @@ func (m *Manager) runAttempt(ctx context.Context, row *rivertype.JobRow, fn func
|
||||
if hasRow {
|
||||
ctx = withJobID(ctx, id)
|
||||
}
|
||||
c := m.counters.Load()
|
||||
if c != nil {
|
||||
c.inFlight.Add(1)
|
||||
}
|
||||
err := m.call(ctx, row, fn)
|
||||
if c != nil {
|
||||
c.inFlight.Add(-1)
|
||||
switch {
|
||||
case err == nil:
|
||||
c.done.Add(1)
|
||||
case c.stopping.Load() && ctx.Err() != nil:
|
||||
c.handedBack.Add(1)
|
||||
default:
|
||||
c.failed.Add(1)
|
||||
}
|
||||
}
|
||||
if err == nil || !hasRow {
|
||||
return err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user