Operator error alerting (#7)

README.md

@@ -115,6 +115,19 @@ records, and (for a laptop with no public IP) a tunnel — that infrastructure
 setup is documented separately in [EMAIL-PIPELINE.md](EMAIL-PIPELINE.md) and
 diagrammed in [email-pipeline-diagram.html](email-pipeline-diagram.html).
 
+To have FileMill email you when it fails systemically (a crashed transformer,
+replies failing for five minutes, intake errors), set `alert_recipient` in
+`config/email.yaml`. Alerts go out through Mailgun from `REPLY_FROM`, and are
+throttled so they can't use up the day's send budget. Check the channel once
+before relying on it:
+
+```powershell
+.\bin\filemill.exe alert-test
+```
+
+It sends one test alert to `alert_recipient`, which counts toward the daily
+alert cap. Run it from a shell that has the Mailgun variables set.
+
 ### Run continuously, in the background
 
 FileMill should run under your Windows account so it can use locally
@@ -140,8 +153,10 @@ in**, and again at logon as a backstop. It launches a **supervisor**
 (`Supervise-FileMill.ps1`) that runs `filemill run` and **restarts it
 automatically if it crashes** — an immediate first retry, then escalating
 backoff (5s, 15s, 30s, 60s, 120s) for repeated rapid failures; a persistent
-crash-loop is logged (alerting is tracked in issue #7). A clean exit (Ctrl+C
-/ shutdown) stops the supervisor.
+crash-loop is logged. When operator alerts are on (`alert_recipient`), the
+restarted worker emails one `restart` alert saying how its predecessor ended,
+throttled so a crash-loop sends one per 15 minutes. A clean exit (Ctrl+C /
+shutdown) stops the supervisor.
 
 Two consequences of running before logon are worth knowing:
 

cmd/filemill/main.go

@@ -13,6 +13,7 @@ import (
 	"syscall"
 	"time"
 
+	"filemill/internal/alert"
 	"filemill/internal/app"
 	"filemill/internal/mailgun"
 )
@@ -66,6 +67,30 @@ func main() {
 			fatal(err)
 		}
 		fmt.Printf("id: %s\noperation: %s\nstatus: %s\nmessage: %s\n", job.ID, job.Operation, job.Status, job.Message)
+	case "alert-test":
+		// Sends one alert to alert_recipient, so the channel is proven (and
+		// its spam placement known) before anything relies on it.
+		if len(os.Args) != 2 {
+			usage()
+			os.Exit(2)
+		}
+		mail, err := mailgun.Load(root, application, log.New(os.Stderr, "mailgun ", log.LstdFlags|log.LUTC))
+		if err != nil {
+			fatal(err)
+		}
+		if mail == nil {
+			fatal(fmt.Errorf("alert-test sends through Mailgun: set MAILGUN_API_KEY, MAILGUN_WEBHOOK_SIGNING_KEY, MAILGUN_DOMAIN, and REPLY_FROM"))
+		}
+		to := mail.AlertRecipient()
+		if to == "" {
+			fatal(fmt.Errorf("no alert_recipient in config/email.yaml"))
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+		defer cancel()
+		if err := alert.SendTest(ctx, mail, application, to, time.Now()); err != nil {
+			fatal(err)
+		}
+		fmt.Printf("test alert sent to %s; check that it arrives and isn't marked as spam\n", to)
 	case "run":
 		once := len(os.Args) == 3 && os.Args[2] == "--once"
 		if len(os.Args) > 2 && !once {
@@ -80,13 +105,35 @@ func main() {
 		defer cancel()
 		var server *http.Server
 		serverErrs := make(chan error, 1)
+		// The alert Emailer runs on a context of its own, stopped only after the
+		// job loop and the webhook server have finished, so an alert raised
+		// while they shut down still goes out. alertsDone closes once it has
+		// drained.
+		alertCtx, stopAlerts := context.WithCancel(context.Background())
+		defer stopAlerts()
+		var alertsDone chan struct{}
+		// reporter takes run's own restart alert: the Emailer once it is wired,
+		// otherwise nothing.
+		var reporter alert.Reporter = alert.Nop{}
 		if !once {
 			mailLog := log.New(io.MultiWriter(os.Stderr, application.LogWriter()), "mailgun ", log.LstdFlags|log.LUTC)
 			mail, err := mailgun.Load(root, application, mailLog)
 			if err != nil {
 				fatal(err)
 			}
 			if mail != nil {
+				// Operator alerts go out through the Mailgun adapter, from
+				// REPLY_FROM, so they exist only when it does. The reporters are
+				// set before any goroutine starts, so nothing reports into the
+				// no-op default by accident.
+				var emailer *alert.Emailer
+				if to := mail.AlertRecipient(); to != "" {
+					alertLog := log.New(io.MultiWriter(os.Stderr, application.LogWriter()), "alert ", log.LstdFlags|log.LUTC)
+					emailer = alert.NewEmailer(mail, application, to, mail.AlertConfig(), time.Now, alertLog)
+					application.SetReporter(emailer)
+					mail.SetReporter(emailer)
+					reporter = emailer
+				}
 				server = &http.Server{Addr: os.Getenv("LISTEN_ADDR"), Handler: mail.Handler()}
 				if server.Addr == "" {
 					server.Addr = ":8080"
@@ -107,6 +154,13 @@ func main() {
 					fatal(fmt.Errorf("webhook server: %w", err))
 				}
 				mailLog.Printf("FileMill %s — webhook listening on %s; delivery loop started", version, server.Addr)
+				if emailer != nil {
+					alertsDone = make(chan struct{})
+					go func() { emailer.Run(alertCtx); close(alertsDone) }()
+					mailLog.Printf("operator alerts: sent to %s", mail.AlertRecipient())
+				} else {
+					mailLog.Print("operator alerts: disabled (no alert_recipient in email.yaml)")
+				}
 				go func() {
 					if err := server.Serve(listener); err != nil && err != http.ErrServerClosed {
 						mailLog.Printf("server: %v", err)
@@ -125,6 +179,21 @@ func main() {
 			// this sweep runs unconditionally in continuous mode rather than
 			// nested under the mailgun branch above.
 			go application.SweepExpiredJobs(ctx)
+
+			// Only a starting worker may conclude that a job left running is
+			// orphaned, and only the continuous one: it holds the webhook port
+			// by now, so a second worker that lost the bind has already exited
+			// without touching the first one's jobs. --once must never do this
+			// — it binds nothing, and `run --once` alongside the real worker
+			// would mark that worker's live job interrupted, which the delivery
+			// loop takes as finished. Other commands never do it either.
+			interrupted, err := application.InterruptLeftoverJobs()
+			if err != nil {
+				fatal(err)
+			}
+			if restart, ok := restartAlert(os.Getenv(previousExitEnv), os.Getenv(rapidRestartsEnv), interrupted); ok {
+				reporter.Report(restart)
+			}
 		}
 		runErr := application.Run(ctx, once)
 		if server != nil {
@@ -134,6 +203,12 @@ func main() {
 			_ = server.Shutdown(shutdownCtx)
 			cancelShutdown()
 		}
+		if alertsDone != nil {
+			// Everything that reports has stopped. The Emailer sends what is
+			// still queued, for up to 5 seconds, then returns.
+			stopAlerts()
+			<-alertsDone
+		}
 		// Checked before runErr: when the listener is what failed, Run returns
 		// nil (it stopped because its context was cancelled), and reporting a
 		// clean stop would hide the actual cause.
@@ -153,7 +228,7 @@ func main() {
 
 func usage() {
 	name := filepath.Base(os.Args[0])
-	fmt.Fprintf(os.Stderr, "Usage:\n  %s run [--once]\n  %s submit <operation> <file>\n  %s jobs get <job-id>\n  %s --version\n", name, name, name, name)
+	fmt.Fprintf(os.Stderr, "Usage:\n  %s run [--once]\n  %s submit <operation> <file>\n  %s jobs get <job-id>\n  %s alert-test\n  %s --version\n", name, name, name, name, name)
 }
 
 func fatal(err error) { fmt.Fprintln(os.Stderr, "filemill:", err); os.Exit(1) }

cmd/filemill/restart.go

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	"filemill/internal/alert"
+)
+
+// The supervisor (scripts/Supervise-FileMill.ps1) sets these for each relaunch,
+// so the new worker can report how the last one ended: a process can't report
+// its own death.
+const (
+	previousExitEnv  = "FILEMILL_PREVIOUS_EXIT"  // the last exit code; unset on a first launch
+	rapidRestartsEnv = "FILEMILL_RAPID_RESTARTS" // restarts in quick succession so far
+)
+
+// crashLoopAt matches the supervisor's $crashLoopAt: this many rapid restarts
+// is a crash-loop.
+const crashLoopAt = 4
+
+// restartAlert builds the restart alert, if there is anything to report: a
+// supervisor relaunch, or jobs the last worker left running, which means it
+// didn't stop cleanly even when no supervisor was there to say so.
+func restartAlert(previousExit, rapidRestarts string, interrupted int) (alert.Alert, bool) {
+	if previousExit == "" && interrupted == 0 {
+		return alert.Alert{}, false
+	}
+	var summary, detail strings.Builder
+	if previousExit != "" {
+		fmt.Fprintf(&summary, "worker restarted after exit code %s", previousExit)
+		fmt.Fprintf(&detail, "The supervisor restarted the FileMill worker after exit code %s.\n", previousExit)
+		if rapid, err := strconv.Atoi(rapidRestarts); err == nil && rapid >= crashLoopAt {
+			fmt.Fprintf(&summary, " (crash-loop: %d rapid restarts)", rapid)
+			fmt.Fprintf(&detail, "It has restarted %d times in quick succession, a crash-loop. The supervisor backs off up to 2 minutes between attempts; data\\logs\\supervisor.log and filemill.log have the details.\n", rapid)
+		}
+		if interrupted > 0 {
+			fmt.Fprintf(&summary, ", %d job(s) interrupted", interrupted)
+		}
+	} else {
+		fmt.Fprintf(&summary, "worker started after an unclean stop: %d job(s) interrupted", interrupted)
+		detail.WriteString("The FileMill worker found jobs its predecessor left running, so the last worker did not stop cleanly: a crash, a kill, or a power loss.\n")
+	}
+	if interrupted > 0 {
+		fmt.Fprintf(&detail, "\n%d job(s) were running when the previous worker stopped. They are marked interrupted and not retried; each sender's reply asks them to send the file again.\n", interrupted)
+	}
+	return alert.Alert{Category: "restart", Summary: summary.String(), Detail: detail.String()}, true
+}

cmd/filemill/restart_test.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"strings"
+	"testing"
+)
+
+// A process can't report its own death, so the restarted worker reports it,
+// from what the supervisor passes it and the jobs its predecessor left
+// running.
+func TestRestartAlert(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		previousExit string // FILEMILL_PREVIOUS_EXIT; empty on a first launch
+		rapid        string // FILEMILL_RAPID_RESTARTS
+		interrupted  int
+		wantReport   bool
+		wantSummary  string
+		wantDetail   []string
+	}{
+		{name: "first launch after a clean stop"},
+		{name: "restart after a crash", previousExit: "2", rapid: "1", wantReport: true,
+			wantSummary: "worker restarted after exit code 2", wantDetail: []string{"exit code 2"}},
+		{name: "below the crash-loop threshold", previousExit: "2", rapid: "3", wantReport: true,
+			wantSummary: "worker restarted after exit code 2"},
+		{name: "crash-loop", previousExit: "2", rapid: "4", wantReport: true,
+			wantSummary: "worker restarted after exit code 2 (crash-loop: 4 rapid restarts)", wantDetail: []string{"crash-loop"}},
+		{name: "unreadable restart count", previousExit: "1", rapid: "lots", wantReport: true,
+			wantSummary: "worker restarted after exit code 1"},
+		// Jobs left running mean the last worker didn't stop cleanly, even
+		// with no supervisor to say so.
+		{name: "interrupted jobs without a supervisor", interrupted: 3, wantReport: true,
+			wantSummary: "worker started after an unclean stop: 3 job(s) interrupted", wantDetail: []string{"3 job(s)"}},
+		{name: "crash with interrupted jobs", previousExit: "-1073741819", rapid: "0", interrupted: 1, wantReport: true,
+			wantSummary: "worker restarted after exit code -1073741819, 1 job(s) interrupted", wantDetail: []string{"1 job(s)"}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			a, ok := restartAlert(tc.previousExit, tc.rapid, tc.interrupted)
+			if ok != tc.wantReport {
+				t.Fatalf("report = %t, want %t (alert %+v)", ok, tc.wantReport, a)
+			}
+			if !ok {
+				return
+			}
+			if a.Category != "restart" || a.Summary != tc.wantSummary {
+				t.Errorf("alert = %s %q, want restart %q", a.Category, a.Summary, tc.wantSummary)
+			}
+			for _, want := range tc.wantDetail {
+				if !strings.Contains(a.Detail, want) {
+					t.Errorf("detail lacks %q:\n%s", want, a.Detail)
+				}
+			}
+		})
+	}
+}

config/email.yaml.example

@@ -26,3 +26,18 @@ max_attachment_bytes: 20971520
 # Timeout for an outbound Mailgun send (reply). Bounds a hung connection so it
 # can't stall the delivery loop. Defaults to 30 if unset or <= 0.
 send_timeout_seconds: 30
+
+# Operator alerts. Systemic failures (a crashed or timed-out transformer, replies
+# failing for 5 minutes, intake errors, retention sweeps that can't delete) are
+# emailed to this address, from REPLY_FROM. Empty or absent turns them off.
+# Check the channel once with `filemill alert-test`, which sends one test alert.
+#
+# Alerts are throttled: one per category every alert_cooldown_minutes (the next
+# email counts the ones held back), and at most alert_max_per_hour and
+# alert_max_per_day across every category. The daily cap is sized for the
+# Mailgun Free plan, whose 100 sends a day are shared with replies; raise it
+# only on a paid plan. The values shown are the defaults.
+alert_recipient: ""   # e.g. support@example.com
+# alert_cooldown_minutes: 15
+# alert_max_per_hour: 10
+# alert_max_per_day: 20

internal/alert/alert.go

@@ -0,0 +1,54 @@
+// Package alert emails the operator when FileMill fails systemically.
+//
+// It is a leaf package because both app and mailgun report, and mailgun
+// imports app, so neither can own it. See ERROR-ALERTING-PLAN.md.
+package alert
+
+import (
+	"context"
+	"time"
+)
+
+// Alert is one systemic failure.
+type Alert struct {
+	Category string // throttle key, e.g. "job-systemic"
+	Summary  string // one line; becomes the subject
+	Detail   string // job id, operation, input name, error, truncated output
+}
+
+// Reporter records a systemic failure. Report must not block or panic.
+type Reporter interface{ Report(Alert) }
+
+// Nop is the default Reporter: alerting disabled.
+type Nop struct{}
+
+func (Nop) Report(Alert) {}
+
+// Mailer sends one plain-text message with its subject verbatim.
+// *mailgun.Service satisfies it.
+type Mailer interface {
+	SendAlert(ctx context.Context, to, subject, text string) error
+}
+
+// Ledger is the persisted throttle state. *store.Store satisfies it.
+//
+// It lives in the database, not in memory, because the supervisor restarts a
+// crashing worker every ≤120s and an in-memory throttle would reset each time.
+type Ledger interface {
+	// LastAlert returns when category last sent and how many of its alerts
+	// have been suppressed since. A zero sentAt means it has never sent; it
+	// can still have a suppressed count, held back by a global cap.
+	LastAlert(category string) (sentAt time.Time, suppressed int, err error)
+	// RecordAlertSent records a send at at. It leaves the suppressed count
+	// alone: a send that then fails reported nothing, so the count must
+	// survive for the next email that does go out.
+	RecordAlertSent(category string, at time.Time) error
+	// RecordAlertSuppressed increments category's suppressed count.
+	RecordAlertSuppressed(category string) error
+	// ClearSuppressed zeroes category's suppressed count, once an email
+	// carrying it has been sent.
+	ClearSuppressed(category string) error
+	// AlertSendsSince returns the time of every send after t, in any
+	// category, oldest first. It must cover at least the last 24h.
+	AlertSendsSince(t time.Time) ([]time.Time, error)
+}