feat: trace and measure everything dispatch does in the background, on forge v1.12.9 and grove v1.8.0

AGENTS.md

@@ -114,7 +114,8 @@ The same file adds four jobs. Native query removal guard compatibility starts a
 | `store/` | Aggregate `Store`, plus `memory`, `sqlite`, `postgres`, `mongo`, `redis` and `storetest` |
 | `ext/` | Opt-in lifecycle hook interfaces |
 | `audit_hook/`, `relay_hook/` | Hooks that send lifecycle events to an audit recorder and to relay |
-| `middleware/`, `observability/`, `backoff/`, `scope/`, `paging/` | Job middleware, OpenTelemetry, retry delays, tenant scope, cursor paging |
+| `middleware/`, `observability/`, `backoff/`, `scope/`, `paging/` | Job middleware, the legacy lifecycle counters, retry delays, tenant scope, cursor paging |
+| `telemetry/` | The spans and labelled metrics every subsystem records through. A leaf: exec must not import it |
 | `api/`, `dwp/`, `client/`, `stream/` | REST handlers, the Dispatch Wire Protocol, the Go client, the event broker |
 | `security/`, `operator/` | Remote operator authentication and authorization, payload-safe durable reads and commands |
 | `extension/` | The Forge extension, with the dashboard contract in `extension/contract/` |

README.md

@@ -137,7 +137,7 @@ Most job queues give you jobs, and you build workflows, schedules and failure ha
 | More than one worker | Worker registration, heartbeats and reassignment of jobs from workers that stop heartbeating, plus a Kubernetes Lease-based cluster store in `cluster/k8s` |
 | Big jobs next to small ones | Resource-aware admission: jobs declare CPU, memory, disk or custom resources and workers only take what fits |
 | Large inputs and outputs | An artifact plane that tracks objects in [Trove](https://github.com/xraph/trove) and stages them to a local cache |
-| Visibility | Logging, recovery, timeout, OpenTelemetry tracing and metrics middleware, plus lifecycle hooks for audit logs and [Relay](https://github.com/xraph/relay) webhooks |
+| Visibility | OpenTelemetry spans and labelled metrics for job attempts, cron, workflows, the durable runtime, the outbox and the artifact sweeper, plus lifecycle hooks for audit logs and [Relay](https://github.com/xraph/relay) webhooks |
 | An admin surface | REST routes for jobs, workflow runs, the DLQ and crons, the Dispatch Wire Protocol over WebSocket, and Go, TypeScript and Python clients |
 
 The [architecture guide](https://www.xraph.com/docs/dispatch/architecture) shows how the dispatcher, the engine, the worker pool and the stores fit together.

artifact/staging/middleware.go

@@ -6,12 +6,15 @@ import (
 	"fmt"
 
 	log "github.com/xraph/go-utils/log"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/xraph/dispatch"
 	"github.com/xraph/dispatch/artifact"
 	"github.com/xraph/dispatch/artifact/cache"
 	"github.com/xraph/dispatch/job"
 	"github.com/xraph/dispatch/middleware"
+	"github.com/xraph/dispatch/telemetry"
 )
 
 // SpecLookup resolves a job name to its declared inputs. The engine
@@ -72,7 +75,15 @@ func Middleware(
 			return verr
 		}
 
-		staged, release, serr := stageInputs(ctx, svc, c, specs, bindings, cfg.Logger)
+		// Nested under the attempt's span when there is one, and absent
+		// when there is not: staging has no Telemetry of its own.
+		stageCtx, span := telemetry.StartChild(ctx, "stage job inputs",
+			trace.WithAttributes(attribute.Int("dispatch.job.inputs", len(bindings))))
+		staged, release, serr := stageInputs(stageCtx, svc, c, specs, bindings, cfg.Logger)
+		if serr != nil {
+			telemetry.Fail(span, "staging failed", telemetry.ErrorType(serr))
+		}
+		span.End()
 
 		// Release before returning whatever happens — including a panic
 		// unwinding through here — or a failed job would pin cache entries

artifact/sweeper/sweeper.go

@@ -8,8 +8,11 @@ import (
 	"time"
 
 	log "github.com/xraph/go-utils/log"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/xraph/dispatch/artifact"
+	"github.com/xraph/dispatch/telemetry"
 )
 
 // Defaults for a sweeper built without options.
@@ -55,6 +58,7 @@ type Sweeper struct {
 	enabled     bool
 	isLeader    func() bool
 	observer    Observer
+	tel         *telemetry.Telemetry
 	stopCh      chan struct{}
 	stopOnce    sync.Once
 	wg          sync.WaitGroup
@@ -108,6 +112,14 @@ func WithObserver(o Observer) Option {
 	return func(s *Sweeper) { s.observer = o }
 }
 
+// WithTelemetry sets where the sweeper records its spans and metrics: a
+// root "sweep artifacts" span per pass that does work, and the swept,
+// purged and bytes reclaimed counters. The Forge extension passes the
+// engine's own.
+func WithTelemetry(t *telemetry.Telemetry) Option {
+	return func(s *Sweeper) { s.tel = t }
+}
+
 // WithLogger sets the logger.
 func WithLogger(l log.Logger) Option {
 	return func(s *Sweeper) { s.logger = l }
@@ -154,12 +166,17 @@ func (s *Sweeper) active() bool {
 // retention has elapsed, then orphans — artifacts with no links at all,
 // which can only result from a partial failure during creation.
 func (s *Sweeper) SweepOnce(ctx context.Context) (Result, error) {
-	var res Result
-
 	if !s.active() {
-		return res, nil
+		return Result{}, nil
 	}
 
+	return s.sweep(ctx)
+}
+
+// sweep is SweepOnce without the leadership check.
+func (s *Sweeper) sweep(ctx context.Context) (Result, error) {
+	var res Result
+
 	eligible, err := s.store.SweepEphemeral(ctx, artifact.SweepOpts{
 		Retention: s.retention,
 		Limit:     s.batchSize,
@@ -169,7 +186,11 @@ func (s *Sweeper) SweepOnce(ctx context.Context) (Result, error) {
 		return res, fmt.Errorf("dispatch/artifact/sweeper: sweep ephemeral: %w", err)
 	}
 
-	res.Swept += s.report(ctx, eligible, "retention")
+	retention := s.report(ctx, eligible, "retention")
+	res.Swept += retention
+	if !s.dryRun {
+		s.tel.Artifacts().Swept("retention", retention)
+	}
 
 	cutoff := time.Now().UTC().Add(-s.orphanGrace())
 
@@ -179,7 +200,9 @@ func (s *Sweeper) SweepOnce(ctx context.Context) (Result, error) {
 			return res, fmt.Errorf("dispatch/artifact/sweeper: sweep orphans: %w", oerr)
 		}
 
-		res.Swept += s.report(ctx, orphans, "orphan")
+		orphaned := s.report(ctx, orphans, "orphan")
+		res.Swept += orphaned
+		s.tel.Artifacts().Swept("orphan", orphaned)
 	}
 
 	return res, nil
@@ -238,12 +261,22 @@ func (s *Sweeper) report(ctx context.Context, artifacts []*artifact.Artifact, re
 // one unreachable object cannot stall reclamation of everything else. The
 // next pass retries it.
 func (s *Sweeper) PurgeOnce(ctx context.Context) (Result, error) {
-	var res Result
-
 	if !s.active() || s.dryRun {
-		return res, nil
+		return Result{}, nil
 	}
 
+	return s.purge(ctx)
+}
+
+// purge is PurgeOnce without the leadership and dry-run checks.
+func (s *Sweeper) purge(ctx context.Context) (Result, error) {
+	var res Result
+	defer func() {
+		s.tel.Artifacts().Purged(telemetry.PurgeOK, res.Purged)
+		s.tel.Artifacts().Purged(telemetry.PurgeSkipped, res.Skipped)
+		s.tel.Artifacts().BytesReclaimed(res.BytesReclaimed)
+	}()
+
 	purgeable, err := s.store.ListPurgeable(ctx, s.purgeGrace, s.batchSize)
 	if err != nil {
 		return res, fmt.Errorf("dispatch/artifact/sweeper: list purgeable: %w", err)
@@ -298,17 +331,44 @@ func (s *Sweeper) PurgeOnce(ctx context.Context) (Result, error) {
 }
 
 // RunOnce performs a sweep followed by a purge.
-func (s *Sweeper) RunOnce(ctx context.Context) (Result, error) {
-	swept, err := s.SweepOnce(ctx)
-	if err != nil {
-		return swept, err
+//
+// A pass that does work (this instance leads and the sweeper is enabled)
+// runs under one "sweep artifacts" span, a child of ctx's span if it has
+// one. The background loop gives it none, so each pass is its own trace.
+// A follower's pass does nothing and starts no span.
+func (s *Sweeper) RunOnce(ctx context.Context) (res Result, err error) {
+	if !s.active() {
+		return Result{}, nil
 	}
 
-	purged, err := s.PurgeOnce(ctx)
+	ctx, span := s.tel.Start(ctx, "sweep artifacts",
+		trace.WithAttributes(attribute.Bool("dispatch.artifact.dry_run", s.dryRun)))
+	defer func() {
+		span.SetAttributes(
+			attribute.Int("dispatch.artifact.swept", res.Swept),
+			attribute.Int("dispatch.artifact.purged", res.Purged),
+			attribute.Int("dispatch.artifact.skipped", res.Skipped),
+			attribute.Int64("dispatch.artifact.bytes_reclaimed", res.BytesReclaimed),
+		)
+		if err != nil {
+			telemetry.Fail(span, "sweep failed", telemetry.ErrorType(err))
+		}
+		span.End()
+	}()
+
+	swept, err := s.sweep(ctx)
 	if err != nil {
 		return swept, err
 	}
 
+	var purged Result
+	if !s.dryRun {
+		purged, err = s.purge(ctx)
+		if err != nil {
+			return swept, err
+		}
+	}
+
 	return Result{
 		Swept:          swept.Swept,
 		Purged:         purged.Purged,
@@ -322,7 +382,10 @@ func (s *Sweeper) Start(ctx context.Context) error {
 	s.startedOnce.Do(func() {
 		s.wg.Add(1)
 
-		go s.loop(ctx)
+		// The loop runs for the life of the process. The span ctx carries
+		// (a Forge extension's boot span) does not: each pass is its own
+		// trace.
+		go s.loop(telemetry.Detach(ctx))
 	})
 
 	return nil

artifact/sweeper/telemetry_test.go

@@ -0,0 +1,65 @@
+package sweeper_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/xraph/dispatch/artifact/sweeper"
+	"github.com/xraph/dispatch/internal/metricstest"
+	"github.com/xraph/dispatch/internal/oteltest"
+	"github.com/xraph/dispatch/telemetry"
+)
+
+func TestSweepPassIsARootSpanAndFollowersStartNone(t *testing.T) {
+	sr := oteltest.Record(t)
+	r := newRig(t)
+	r.createEphemeral(t, newOwner(), "temp.bin", 48*time.Hour)
+	f := metricstest.New()
+
+	leader := false
+	s := sweeper.New(r.store, r.backend,
+		sweeper.WithRetention(0),
+		sweeper.WithPurgeGrace(0),
+		sweeper.WithInterval(5*time.Millisecond),
+		sweeper.WithLeaderCheck(func() bool { return leader }),
+		sweeper.WithTelemetry(telemetry.New(telemetry.WithMetrics(f))))
+
+	if _, err := s.RunOnce(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if spans := sr.Ended(); len(spans) != 0 {
+		t.Fatalf("a follower's pass started %d spans", len(spans))
+	}
+
+	leader = true
+	// Started under a context carrying a span, as an extension's Start is.
+	if err := s.Start(oteltest.SpanContext(t)); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = s.Stop(context.Background()) })
+	pass := sr.WaitFor(t, "sweep artifacts", 1, 5*time.Second)[0]
+	if err := s.Stop(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+
+	if !oteltest.IsRoot(pass) {
+		t.Error("a background sweep pass should start its own trace")
+	}
+	if got, _ := oteltest.Attr(pass, "dispatch.artifact.swept"); got != "1" {
+		t.Errorf("swept = %q", got)
+	}
+	if v, ok := f.CounterValue(telemetry.MetricArtifactsSwept, "reason", "retention"); !ok || v != 1 {
+		t.Errorf("swept metric = %v, %v", v, ok)
+	}
+	// A zero purge grace purges in the same pass that swept.
+	if got, _ := oteltest.Attr(pass, "dispatch.artifact.purged"); got != "1" {
+		t.Errorf("purged = %q", got)
+	}
+	if v, ok := f.CounterValue(telemetry.MetricArtifactsPurged, "result", telemetry.PurgeOK); !ok || v != 1 {
+		t.Errorf("purged metric = %v, %v", v, ok)
+	}
+	if v, ok := f.CounterValue(telemetry.MetricArtifactBytesReclaimed); !ok || v != float64(len("payload")) {
+		t.Errorf("bytes reclaimed = %v, %v", v, ok)
+	}
+}