Harden refresh/runtime flows and add scheduler observability

Tighten TargetURL fallback behavior, wire filerepo refreshes into ServiceHealth with proper skip semantics, and clean up App.Open failure shutdown paths. Add dedicated Prometheus metrics for bus delivery and scheduler lifecycle/execution/restore state, plus coverage for the new observability and concurrency paths.

Author: ExplodingDragon

.gitignore

@@ -29,4 +29,5 @@ config.local.yaml
 web/node_modules/
 web/.angular/
 web/dist/
-node_modules
+node_modules
+/.gocache

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/google/go-containerregistry v0.21.7
 	github.com/klauspost/compress v1.18.6
 	github.com/prometheus/client_golang v1.23.2
+	github.com/prometheus/client_model v0.6.2
 	github.com/spf13/afero v1.15.0
 	github.com/stretchr/testify v1.11.1
 	github.com/ulikunitz/xz v0.5.15
@@ -37,7 +38,6 @@ require (
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/pjbgf/sha1cd v0.6.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.69.0 // indirect
 	github.com/prometheus/procfs v0.21.0 // indirect
 	github.com/sergi/go-diff v1.4.0 // indirect

pkg/app/app.go

@@ -140,8 +140,8 @@ func Validate(doc *config.Document) error {
 	stats := httpcache.NewStats(registry)
 	downloads := httpcache.NewDownloadLimiter(copy.Storage.Download.MaxActive, copy.Storage.Download.MaxActivePerInstance)
 
-	b := bus.New()
-	sched := scheduler.New(b, store)
+	b := bus.NewWithRegisterer(registry)
+	sched := scheduler.New(b, store, registry)
 	validateCtx, validateCancel := context.WithCancel(context.Background())
 	sched.Start(validateCtx)
 	defer validateCancel()
@@ -172,17 +172,22 @@ func Open(ctx context.Context, doc *config.Document, configPath string) (*App, e
 	stats := httpcache.NewStats(metricsReg)
 	downloads := httpcache.NewDownloadLimiter(doc.Storage.Download.MaxActive, doc.Storage.Download.MaxActivePerInstance)
 
-	b := bus.New()
-	sched := scheduler.New(b, store)
+	b := bus.NewWithRegisterer(metricsReg)
+	sched := scheduler.New(b, store, metricsReg)
 
 	lifecycleCtx, stopRuntime := context.WithCancel(context.Background())
 	sched.Start(lifecycleCtx)
+	cleanupOpenFailure := func() {
+		stopRuntime()
+		stopCtx, cancel := context.WithTimeout(context.Background(), drainTimeout)
+		defer cancel()
+		_ = sched.Stop(stopCtx)
+		_ = store.Close()
+	}
 
 	entries, err := planEntries(ctx, doc, store, stats, downloads, sched, b)
 	if err != nil {
-		sched.Stop(ctx)
-		stopRuntime()
-		_ = store.Close()
+		cleanupOpenFailure()
 		return nil, err
 	}
 
@@ -204,7 +209,7 @@ func Open(ctx context.Context, doc *config.Document, configPath string) (*App, e
 		stopRuntime:   stopRuntime,
 	}
 	if err := app.prepareHandlers(lifecycleCtx); err != nil {
-		_ = store.Close()
+		cleanupOpenFailure()
 		return nil, err
 	}
 

pkg/app/app_plan.go

@@ -20,9 +20,11 @@ const DefaultGCInterval = 24 * time.Hour
 const DefaultMaxActiveDownloads = 64
 const DefaultMaxActiveDownloadsPerInstance = 8
 
+var driverSet = builtinDrivers
+
 func planEntries(ctx context.Context, doc *config.Document, store *blobfs.Store, stats *httpcache.Stats, downloads *httpcache.DownloadLimiter, sched *scheduler.Scheduler, b *bus.Bus) (map[string]*proxyruntime.Entry, error) {
 	plan := proxyruntime.NewPlanContext(store, stats, downloads, doc.Server.Bind, doc.Metrics.Path, sched, b)
-	drivers := builtinDrivers()
+	drivers := driverSet()
 	for _, decl := range doc.Instances {
 		selected, err := decl.SelectMode()
 		if err != nil {

pkg/app/app_test.go

@@ -18,6 +18,7 @@ import (
 	"gopkg.d7z.net/cache-proxy/pkg/config"
 	"gopkg.d7z.net/cache-proxy/pkg/proxy/file"
 	proxyruntime "gopkg.d7z.net/cache-proxy/pkg/runtime"
+	"gopkg.d7z.net/cache-proxy/pkg/scheduler"
 )
 
 func TestValidateRejectsConflictingPaths(t *testing.T) {
@@ -71,7 +72,6 @@ func TestFileProxyCachesImmutableObjects(t *testing.T) {
 	require.Equal(t, int64(1), upstreamRequests.Load())
 }
 
-
 func TestMetricsRequireBearerToken(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
@@ -247,7 +247,6 @@ instances:
 	require.ErrorContains(t, err, "field default_polciy not found")
 }
 
-
 func TestAppCloseRespectsContextWhenHandlerStopBlocks(t *testing.T) {
 	app := &App{
 		stopRuntime: func() {},
@@ -419,6 +418,29 @@ func TestBindHomePageHeadReturnsOK(t *testing.T) {
 	require.Equal(t, "text/html; charset=utf-8", rec.Header().Get("Content-Type"))
 }
 
+func TestOpenStopsSchedulerWhenPrepareHandlersFails(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	var runs atomic.Int32
+	prev := driverSet
+	driverSet = func() map[string]proxyruntime.ModeDriver {
+		drivers := prev()
+		drivers[config.ModeFile] = startFailingDriver{runs: &runs}
+		return drivers
+	}
+	defer func() { driverSet = prev }()
+
+	doc := testDocument(t.TempDir(), []config.Instance{
+		fileInstance(t, "files", "/files", "https://example.com", file.Policy{}),
+	})
+
+	_, err := Open(ctx, doc, "")
+	require.ErrorContains(t, err, "boom")
+	first := runs.Load()
+	time.Sleep(200 * time.Millisecond)
+	require.Equal(t, first, runs.Load())
+}
 
 func openApp(t *testing.T, ctx context.Context, doc *config.Document) *App {
 	return openAppWithConfig(t, ctx, doc, "")
@@ -594,3 +616,25 @@ func (s *cleanupContextInstance) Stop(context.Context) error {
 	s.stopped.Store(true)
 	return s.ctx.Err()
 }
+
+type startFailingDriver struct{ runs *atomic.Int32 }
+
+func (startFailingDriver) Mode() string { return config.ModeFile }
+
+func (d startFailingDriver) Plan(_ context.Context, plan *proxyruntime.InstancePlan) error {
+	plan.Scheduler().Register(scheduler.TaskDef{
+		Key:      scheduler.NewTaskKey(plan.Name(), scheduler.TypeExpireCleanup, ""),
+		Interval: 10 * time.Millisecond,
+		Handler: func(context.Context) error {
+			if d.runs != nil {
+				d.runs.Add(1)
+			}
+			return nil
+		},
+	})
+	return plan.BindPath("/files", config.Expiration(time.Hour), startContextInstance{
+		onStart: func(context.Context) error {
+			return fmt.Errorf("boom")
+		},
+	})
+}

pkg/bus/bus.go

@@ -4,6 +4,8 @@ import (
 	"log/slog"
 	"sync"
 	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 type EventType string
@@ -32,10 +34,15 @@ type MetadataRemovedPayload struct {
 type Bus struct {
 	mu   sync.RWMutex
 	subs map[EventType][]chan Event
+	m    *metrics
 }
 
 func New() *Bus {
-	return &Bus{subs: map[EventType][]chan Event{}}
+	return NewWithRegisterer(nil)
+}
+
+func NewWithRegisterer(reg prometheus.Registerer) *Bus {
+	return &Bus{subs: map[EventType][]chan Event{}, m: newMetrics(reg)}
 }
 
 func (b *Bus) Subscribe(types ...EventType) <-chan Event {
@@ -44,6 +51,9 @@ func (b *Bus) Subscribe(types ...EventType) <-chan Event {
 	defer b.mu.Unlock()
 	for _, t := range types {
 		b.subs[t] = append(b.subs[t], ch)
+		if b.m != nil {
+			b.m.subscribers.WithLabelValues(string(t)).Set(float64(len(b.subs[t])))
+		}
 	}
 	return ch
 }
@@ -52,11 +62,30 @@ func (b *Bus) Publish(evt Event) {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
 	evt.Timestamp = time.Now()
-	for _, ch := range b.subs[evt.Type] {
+	eventType := string(evt.Type)
+	if b.m != nil {
+		b.m.published.WithLabelValues(eventType).Inc()
+	}
+	subs := b.subs[evt.Type]
+	if len(subs) == 0 {
+		if b.m != nil {
+			b.m.dropped.WithLabelValues(eventType, "no_subscriber").Inc()
+		}
+		return
+	}
+	delivered := 0
+	for _, ch := range subs {
 		select {
 		case ch <- evt:
+			delivered++
 		default:
 			slog.Debug("bus event dropped", "type", evt.Type, "reason", "subscriber full")
+			if b.m != nil {
+				b.m.dropped.WithLabelValues(eventType, "subscriber_full").Inc()
+			}
 		}
 	}
+	if b.m != nil {
+		b.m.delivered.WithLabelValues(eventType).Add(float64(delivered))
+	}
 }

pkg/bus/bus_test.go

@@ -4,6 +4,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/require"
 )
 
@@ -165,3 +167,30 @@ func TestBusSameTypeMultipleSubscribe(t *testing.T) {
 	b.Subscribe(EventMetadataDiscovered)
 	b.Publish(Event{Type: EventMetadataDiscovered, Payload: MetadataDiscoveredPayload{Instance: "x"}})
 }
+
+func TestBusMetrics(t *testing.T) {
+	reg := prometheus.NewRegistry()
+	b := NewWithRegisterer(reg)
+	b.Subscribe(EventMetadataDiscovered)
+	b.Publish(Event{Type: EventMetadataDiscovered, Payload: MetadataDiscoveredPayload{Instance: "x"}})
+	b.Publish(Event{Type: EventMetadataRemoved, Payload: MetadataRemovedPayload{Instance: "x"}})
+
+	require.Equal(t, float64(1), metricValue(t, b.m.published.WithLabelValues(string(EventMetadataDiscovered))))
+	require.Equal(t, float64(1), metricValue(t, b.m.delivered.WithLabelValues(string(EventMetadataDiscovered))))
+	require.Equal(t, float64(1), metricValue(t, b.m.subscribers.WithLabelValues(string(EventMetadataDiscovered))))
+	require.Equal(t, float64(1), metricValue(t, b.m.dropped.WithLabelValues(string(EventMetadataRemoved), "no_subscriber")))
+}
+
+func metricValue(t *testing.T, metric prometheus.Metric) float64 {
+	t.Helper()
+	var pb dto.Metric
+	require.NoError(t, metric.Write(&pb))
+	if pb.Counter != nil {
+		return pb.Counter.GetValue()
+	}
+	if pb.Gauge != nil {
+		return pb.Gauge.GetValue()
+	}
+	t.Fatal("unsupported metric type")
+	return 0
+}

pkg/bus/metrics.go

@@ -0,0 +1,36 @@
+package bus
+
+import "github.com/prometheus/client_golang/prometheus"
+
+type metrics struct {
+	published   *prometheus.CounterVec
+	delivered   *prometheus.CounterVec
+	dropped     *prometheus.CounterVec
+	subscribers *prometheus.GaugeVec
+}
+
+func newMetrics(reg prometheus.Registerer) *metrics {
+	if reg == nil {
+		return nil
+	}
+	m := &metrics{
+		published: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "cache_proxy_bus_events_published_total",
+			Help: "Total bus events published by type.",
+		}, []string{"event_type"}),
+		delivered: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "cache_proxy_bus_events_delivered_total",
+			Help: "Total bus event deliveries to subscribers by type.",
+		}, []string{"event_type"}),
+		dropped: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "cache_proxy_bus_events_dropped_total",
+			Help: "Total bus events dropped by type and reason.",
+		}, []string{"event_type", "reason"}),
+		subscribers: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "cache_proxy_bus_subscribers",
+			Help: "Current bus subscriber count by event type.",
+		}, []string{"event_type"}),
+	}
+	reg.MustRegister(m.published, m.delivered, m.dropped, m.subscribers)
+	return m
+}

pkg/proxy/deb/metadata_test.go

@@ -155,9 +155,9 @@ func TestMetadataRequestInReleasePublishesArtifacts(t *testing.T) {
 	}()
 
 	handler.AddRoot("dists/bookworm", []filerepo.MetadataTarget{{
-		URL: "dists/bookworm/InRelease",
+		URL:        "dists/bookworm/InRelease",
 		Candidates: []string{"dists/bookworm/Release"},
-		Kind: "release",
+		Kind:       "release",
 	}})
 	require.NoError(t, handler.RefreshSubPath(ctx, "dists/bookworm"))
 
@@ -214,9 +214,9 @@ func TestMetadataRequestInReleaseRedirectsToReleaseFallback(t *testing.T) {
 	}()
 
 	handler.AddRoot("dists/bookworm", []filerepo.MetadataTarget{{
-		URL: "dists/bookworm/InRelease",
+		URL:        "dists/bookworm/InRelease",
 		Candidates: []string{"dists/bookworm/Release"},
-		Kind: "release",
+		Kind:       "release",
 	}})
 	require.NoError(t, handler.RefreshSubPath(ctx, "dists/bookworm"))
 
@@ -273,9 +273,9 @@ func TestMetadataRequestStartsAsyncRefreshAndReturnsUnavailableUntilReady(t *tes
 	}()
 
 	handler.AddRoot("dists/bookworm", []filerepo.MetadataTarget{{
-		URL: "dists/bookworm/InRelease",
+		URL:        "dists/bookworm/InRelease",
 		Candidates: []string{"dists/bookworm/Release"},
-		Kind: "release",
+		Kind:       "release",
 	}})
 	refreshDone := make(chan struct{})
 	go func() {
@@ -328,9 +328,9 @@ func TestBuildSnapshotRejectsReleaseWithoutPackageIndexes(t *testing.T) {
 		nil,
 	)
 	handler.AddRoot("dists/bookworm", []filerepo.MetadataTarget{{
-		URL: "dists/bookworm/InRelease",
+		URL:        "dists/bookworm/InRelease",
 		Candidates: []string{"dists/bookworm/Release"},
-		Kind: "release",
+		Kind:       "release",
 	}})
 	err = handler.RefreshSubPath(ctx, "dists/bookworm")
 	require.ErrorContains(t, err, "Release contains no package indexes")