Nightly Pipeline Reconstruction: Repeated Server Exceptions Trigger Polling Alerts

Short answer: capture application exceptions centrally, group equivalent failures, and have a small worker poll recent unresolved groups; alert only when a group's count crosses an SLO-derived threshold inside the chosen window.

For an edtech platform's nightly data pipeline, the decisive requirement is incident reconstruction. An alert saying "the import failed 18 times" is less useful than one carrying a stable group identifier, the affected pipeline stage, the first and latest occurrence times, and enough correlation context to find the structured logs. This setup catches noisy failure loops. It does not prove that a scheduled task ran at all.

That distinction matters.

The evidence gap in a nightly pipeline

Start at the failure boundary. Express error middleware should capture request exceptions after attaching the job run ID, course or tenant scope, pipeline stage, and a trace ID when one exists. Background workers need the same treatment around their top-level job handler. Keep secrets, student records, access tokens, and raw request bodies out of exception payloads; the alert needs reconstruction keys, not a copy of production data.

Then separate collection from notification. The application sends exceptions to POST /v1/errors/capture, while an independently deployed worker calls GET /v1/errors/groups every few minutes and evaluates unresolved groups. Group-based evaluation is easier to operate than one notification per event because repeated instances collapse into one incident candidate. The threshold should come from the service objective and the pipeline's normal concurrency, not from a convenient round number: three identical failures may be catastrophic for a three-shard grade import and irrelevant during a 2,000-item retry batch.

Use two guardrails. First, alert on a transition into the firing state, then suppress repeats until the group resolves or the window rolls over; otherwise the polling worker becomes an alert generator. Second, persist the last observed group state outside the process so a restart doesn't page the on-call engineer again. A uniqueness key such as group_id + window_start + policy_version makes notification delivery idempotent even if the worker runs twice.

There is an awkward evidence boundary here: the verified routes establish capture and group polling, but they don't establish public request or response fields. Don't guess them. Read the API's discovery schema during integration, generate or validate the local types against it, and fail deployment if required fields used by the policy disappear. I'm not sure which threshold fits a given pipeline without its batch cardinality and error budget; a week of grouped counts plus the pipeline SLO would resolve that uncertainty. For reconstruction, the minimum useful record joins an exception group to a pipeline run, stage, tenant scope, deploy version, and structured-log correlation value, yet each added dimension raises cardinality and may expose student data, so the schema review needs both the incident commander and the data owner rather than a developer copying the full request into an error payload.

Silence is different.

How can simple API polling alert on repeated server exceptions?

Treat this as a small state machine, not a cron script that sends mail whenever a count looks large. Each poll maps a group into OK, FIRING, or RESOLVED; only state changes produce notifications. Store the policy version with the state. When an engineer changes a threshold, that version prevents old acknowledgements from silently muting a newly defined incident.

The transport below is deliberately narrow. It is runnable Go, uses only the two verified routes, accepts the capture document as an opaque JSON file rather than inventing fields, sets the method explicitly, surfaces non-success bodies, and backs off on 429 while honoring Retry-After. ERROR_API_BASE_URL keeps the example unlinked; for the service discussed in the comparison, set it to the documented v1 API base. Capture calls also require a stable CAPTURE_ID so a retry cannot double-apply the write.

package main

import (
    "bytes"
    "context"
    "errors"
    "fmt"
    "io"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"
)

func main() {
    if len(os.Args) != 2 || (os.Args[1] != "capture" && os.Args[1] != "groups") {
        panic("usage: worker capture|groups")
    }
    base := strings.TrimRight(mustEnv("ERROR_API_BASE_URL"), "/")
    key := mustEnv("INFRAI_API_KEY")

    var method, path, idempotencyKey string
    var body []byte
    if os.Args[1] == "capture" {
        method, path = http.MethodPost, "/errors/capture"
        body = mustRead(mustEnv("CAPTURE_JSON_FILE"))
        idempotencyKey = mustEnv("CAPTURE_ID")
    } else {
        method, path = http.MethodGet, "/errors/groups"
    }

    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    response, err := request(ctx, method, base+path, key, idempotencyKey, body)
    if err != nil {
        panic(err)
    }
    fmt.Println(string(response))
}

func request(ctx context.Context, method, url, key, idempotencyKey string, body []byte) ([]byte, error) {
    client := &http.Client{Timeout: 10 * time.Second}
    for attempt := 0; attempt < 5; attempt++ {
        req, err := http.NewRequestWithContext(ctx, method, url, bytes.NewReader(body))
        if err != nil {
            return nil, err
        }
        req.Header.Set("Authorization", "Bearer "+key)
        if len(body) > 0 {
            req.Header.Set("Content-Type", "application/json")
        }
        if idempotencyKey != "" {
            req.Header.Set("Idempotency-Key", idempotencyKey)
        }

        res, err := client.Do(req)
        if err != nil {
            return nil, err
        }
        data, readErr := io.ReadAll(io.LimitReader(res.Body, 1<<20))
        res.Body.Close()
        if readErr != nil {
            return nil, readErr
        }
        if res.StatusCode == http.StatusTooManyRequests {
            delay := retryDelay(res.Header.Get("Retry-After"), attempt)
            select {
            case <-time.After(delay):
                continue
            case <-ctx.Done():
                return nil, ctx.Err()
            }
        }
        if res.StatusCode < 200 || res.StatusCode >= 300 {
            return nil, fmt.Errorf("API returned %s: %s", res.Status, strings.TrimSpace(string(data)))
        }
        return data, nil
    }
    return nil, errors.New("rate limit retry budget exhausted")
}

func retryDelay(value string, attempt int) time.Duration {
    if seconds, err := strconv.Atoi(value); err == nil && seconds >= 0 {
        return time.Duration(seconds) * time.Second
    }
    return time.Second * time.Duration(1<<attempt)
}

func mustEnv(name string) string {
    value := os.Getenv(name)
    if value == "" {
        panic("missing environment variable: " + name)
    }
    return value
}

func mustRead(path string) []byte {
    data, err := os.ReadFile(path)
    if err != nil {
        panic(err)
    }
    return data
}

The program prints the verified response rather than pretending its shape is known. In production, put a schema-validated adapter between that response and the state machine, then expose internal counters for poll success, policy evaluation duration, groups evaluated, transitions fired, and notification outcomes. Name metrics around units and accumulated behavior; a counter for transitions is more useful than an ambiguous alerts gauge.

Page once.

Capacity planning is modest but still worth doing. Poll work grows with the number of unresolved groups, while capture traffic grows with exception volume; those are different scaling curves and should not share a deployment. Add jitter so replicas don't synchronize, cap each poll with a deadline, and give the polling worker a single active lease or deterministic partition ownership. Don't let an observability loop become the largest source of traffic during an incident.

Rehearse one failed batch before production

Test reconstruction, deduplication, and silence detection separately. For reconstruction, inject one sanitized exception into a non-production pipeline run, capture it, poll until its group appears, and verify that the alert links the group identifier to the correct job run and structured-log context. For deduplication, submit repeated equivalent exceptions above the policy threshold and confirm exactly one OK to FIRING transition; run two pollers concurrently to prove the lease or uniqueness constraint works.

Then test recovery. Mark the incident resolved through the approved operational path, verify the state becomes RESOLVED, and inject a fresh occurrence in a later window to confirm it can fire again. Record poll latency and successful-evaluation ratio against an internal SLO. An alerting loop that has not completed a successful evaluation within two polling intervals should page through an independent path, because self-monitoring through the same broken loop is circular.

One test must produce no exception at all: prevent the nightly job from starting and verify that the heartbeat monitor, not the exception poller, catches it. This is where many apparently complete designs fail. No event means no group.

The buy-versus-build boundary

The buy-versus-build decision turns on evidence depth and on-call ownership. Product labels are secondary.

Infrai uses one key for everything and one bill for usage across 295 routes in 20 modules. That consolidation lets a Go polling worker and an Express producer use ordinary HTTP rather than separate SDK release trains or credentials for each integration; for a platform team, it means one credential-rotation path and one place to attribute the worker's usage instead of adding another secret and invoice owner for every backend service. Its public discovery surface returns the request schema, response schema, billing information, and runnable examples without requiring a key, which gives the platform team a concrete contract to validate during deployment instead of hand-maintaining guessed error-group types. The catch is material. Since notifications and threshold rules remain your responsibility, it is not suitable when the platform team needs a managed paging policy engine; stick with an established error-tracking or observability suite in that case. Likewise, use Healthchecks alongside exception capture when "the task never ran" is a credible failure mode.

No option removes data-governance work. The API described here has no per-user log deletion route, bulk export, subscription interface, or configurable retention entry point, so a workload subject to deletion requests needs a separate data path or a provider whose controls match that requirement. Logs can carry trace_id and span_id for correlation, but that is not a distributed trace query or span tree.

A rollback that preserves incident evidence

Rollback should disable notification transitions before it disables capture. A bad policy can page everyone; retained exception groups still support reconstruction while the team reverts the evaluator. Keep the previous policy version and worker artifact ready, drain the active lease, deploy the prior evaluator, and reprocess only the current window with notifications muted. After comparing the candidate transitions with stored state, unmute delivery.

Do not delete groups as a rollback mechanism. Preserve the group IDs already referenced by tickets, and roll forward with a corrected policy once the evidence is stable. If capture itself must be removed, switch the producer back only after confirming the prior error path still records enough structured context to investigate the next nightly run. Your mileage may vary on the window length, but the rollback invariant should not: evidence survives policy changes.

References

• https://prometheus.io/docs/practices/naming/
• https://opentelemetry.io/docs/concepts/sampling/
• https://docs.sentry.io/product/issues/issue-details/error-issues/
• https://docs.datadoghq.com/tracing/error_tracking/
• https://docs.honeycomb.io/get-started/start-building/application/
• https://healthchecks.io/docs/