Documentation
Guides

SweepScheduler

SweepScheduler

SweepScheduler runs the relation graph's structural sweep on a cron schedule. It is part of smeldr.dev/agent — import it alongside your agent configuration.

The structural sweep is Layer 3 of the Content Relations correctness model: it catches stale edges that Layer 1 (save-path sync) and Layer 2 (AfterRelationCascade) cannot reach — bulk imports, database restores, manual deletions, signals missed during a restart. See Content Relations for the full three-layer model.

SweepFunc

SweepFunc is the function type that SweepScheduler calls on each tick:

type SweepFunc func(ctx context.Context) (flagged, skipped int, err error)
  • flagged — number of edges whose target was found not alive (stamped with invalid_at)
  • skipped — number of edges where the liveness check failed (edge left unchanged)
  • err — returned only for fatal errors that prevented the sweep from running at all

app.SweepStructural satisfies SweepFunc directly — no adapter needed.

Setup

import agentpkg "smeldr.dev/agent"

sweep, err := agentpkg.NewSweepScheduler(
    "0 * * * *",           // 5-field cron: every hour at :00
    "Europe/Copenhagen",   // IANA timezone; empty string defaults to UTC
    app.SweepStructural,   // satisfies SweepFunc directly
)
if err != nil {
    log.Fatal(err)
}

sweep.Start()
defer sweep.Stop()

The timezone is validated at construction time — NewSweepScheduler returns an error if the location string is unrecognised. The schedule is a standard 5-field cron expression (minute hour day month weekday).

Start and Stop

sweep.Start() activates the scheduler. The first tick fires according to the cron expression — no immediate run on startup.

sweep.Stop() shuts the scheduler down and waits for any in-flight sweep to finish before returning. Call it in a defer at startup or in your shutdown handler to avoid cutting off a sweep mid-run.

Singleton mode

SweepScheduler uses LimitModeReschedule (gocron's singleton mode). If a sweep is still running when the next cron tick arrives, the tick is rescheduled rather than starting a second concurrent sweep. This prevents overlapping sweeps from competing for database connections or producing conflicting invalid_at stamps.

App.SweepStructural vs. store.SweepStructural

Two implementations of SweepFunc are available, with different trade-offs:

app.SweepStructural — the high-level variant.

  • Uses a built-in TargetChecker that queries smeldr_dynamic_content by ID

and checks status = 'published'. Covers all runtime-defined content types (those registered via DefineContentType / smeldr_dynamic_content).

  • For each stale edge, fires AfterRelationCascade via the signal bus — the

same path as Layer 2. Handlers registered with App.OnSignal receive the notification without knowing whether it came from a reactive cascade or a sweep.

  • Does not cover compiled content types (those registered via Module[T]).
flagged, skipped, err := app.SweepStructural(ctx)

store.SweepStructural — the low-level variant.

  • Accepts a custom TargetChecker and an onStale callback. Use this when:

- Your application uses compiled Module[T] content types and needs to check their liveness tables. - You want to hard-delete stale edges immediately rather than stamping invalid_at. - You need to route stale-edge handling separately from the signal bus.

check := func(ctx context.Context, targetType, targetID string) (bool, error) {
    // query your own tables here
    alive, err := myRepo.Exists(ctx, targetID)
    return alive, err
}

flagged, skipped, err := store.SweepStructural(ctx, check, func(ctx context.Context, e smeldr.RelationEdge) {
    // called once per stale edge
    // edge has already been stamped with invalid_at = now
    _ = store.Delete(ctx, e.ID)       // hard-delete if desired
    reviewQueue.Enqueue(e.SourceType, e.SourceID)
})

If you use both runtime-defined and compiled content types, combine them: pass a TargetChecker to store.SweepStructural that queries both smeldr_dynamic_content and your compiled-type tables.

TargetChecker

type TargetChecker func(ctx context.Context, targetType, targetID string) (alive bool, err error)

Called once per unique target (not once per edge). If a target has 50 source edges pointing at it, the checker is called once.

  • alive = true — target exists and is in a live state; edges are left unchanged.
  • alive = false — target is gone or no longer live; all source edges for this

target are stamped with invalid_at = now.

  • err != nil — the check could not be performed (database error, timeout). The

target's edges are skipped; the skipped counter is incremented. The sweep continues with the next target.

onStale callback

The onStale callback is called once per stale edge, after invalid_at has been stamped:

func(ctx context.Context, e smeldr.RelationEdge)

When using app.SweepStructural, the built-in onStale fires AfterRelationCascade for the source item. When using store.SweepStructural, your callback receives the edge directly.

Common patterns for the callback:

  • Queue a review job so an editor can decide what to do with the dependent item.
  • Hard-delete the edge via store.Delete(ctx, e.ID) if history is not needed.
  • Set a derived field on the dependent item (e.g. has_stale_relation = true)

so a dashboard can surface items needing attention.

  • Trigger an agent to propose a replacement target.

Logging

SweepScheduler logs via the application's slog handler:

  • Debug — logged on every tick when flagged == 0 and skipped == 0. No noise in normal operation.
  • Info — logged when flagged > 0 or skipped > 0, including the counts. A non-zero flagged count indicates that something changed outside the normal save path since the last sweep.

Treat a non-zero flagged count after a run as a signal worth investigating: it means a bulk import bypassed the save hook, a restore happened, or a direct database deletion was made.

Why invalid_at, not delete

Stale edges are stamped with invalid_at rather than deleted. This preserves the history of the relation graph — an audit trail, a dashboard showing that an item was once depended on by others, or a governance workflow that reviews stale edges before removing them all need the edge records to persist.

invalid_at takes the edge out of the active set (queries filter it out) without destroying the record. Hard delete is always available:

_ = store.Delete(ctx, edgeID)

Use hard delete when storage hygiene matters more than history, or in onStale callbacks where you know the edge will never be relevant again.

Timezone note

The timezone parameter is an IANA location string. Empty string defaults to UTC.

If your sweep schedule is time-of-day sensitive — a daily sweep at midnight local time, or an hourly sweep that should align with business hours — specify the location explicitly:

// Daily at midnight Copenhagen time, not UTC midnight
agentpkg.NewSweepScheduler("0 0 * * *", "Europe/Copenhagen", app.SweepStructural)

On Alpine or scratch containers, import the embedded timezone database to ensure time.LoadLocation works without an OS tzdata package:

import _ "time/tzdata"

smeldr.dev/agent embeds time/tzdata since v0.5.0 — you do not need this import if you are on v0.5.0 or later.