feat(observe): expose diagnostic lifecycle expiry

config/observability-dashboards.yaml

@@ -64,6 +64,37 @@ dashboards:
         query: max(gha_fleet_queue_uncovered_running)
         unit: count
         description: GitHub running jobs absent from the durable queue model.
+  - id: diagnostic_storage
+    title: Diagnostic storage retention
+    refresh_seconds: 60
+    default_range: 24h
+    owner: fleet-operations
+    runbook: https://github.com/NDDev-OpenNetwork/github-actions/blob/main/docs/runbooks/fleet-alerts.md
+    panels:
+      - id: expiration_eligible
+        title: Objects past lifecycle eligibility
+        kind: timeseries
+        query: max(gha_diagnostic_storage_expiration_eligible_objects)
+        unit: count
+        description: Objects retained after their S3 UTC-midnight expiration boundary.
+      - id: next_expiration
+        title: Time to next lifecycle boundary
+        kind: timeseries
+        query: max(gha_diagnostic_storage_next_expiration_seconds)
+        unit: seconds
+        description: Seconds until the earliest retained object becomes expiration eligible.
+      - id: objects
+        title: Retained diagnostic objects
+        kind: timeseries
+        query: max(gha_diagnostic_storage_objects)
+        unit: count
+        description: Current object population under the managed diagnostic prefix.
+      - id: oldest_object_age
+        title: Oldest retained object age
+        kind: timeseries
+        query: max(gha_diagnostic_storage_oldest_object_age_seconds)
+        unit: seconds
+        description: Age of the oldest retained diagnostic object.
   - id: lifecycle_latency
     title: Job lifecycle latency
     refresh_seconds: 30

config/observability-rules.yaml

@@ -34,6 +34,22 @@ rules:
     summary: Durable diagnostic export is failing.
     action: Preserve the spool and restore RustFS routing or credentials without pruning evidence.
     recovery: Consecutive failures and pending backlog return to zero after an exact successful export.
+  - id: diagnostic_retention_overdue
+    severity: ticket
+    query_language: promql
+    stream_name: gha_diagnostic_storage_expiration_eligible_objects
+    expression: max(gha_diagnostic_storage_expiration_eligible_objects)
+    operator: ">"
+    threshold: 0
+    evaluation_seconds: 300
+    hold_seconds: 3600
+    destination_ref: fleet_oncall
+    enabled: false
+    owner: fleet-operations
+    runbook: https://github.com/NDDev-OpenNetwork/github-actions/blob/main/docs/runbooks/fleet-alerts.md
+    summary: Diagnostic objects remain after their S3 lifecycle eligibility boundary.
+    action: Preserve object-age counts and lifecycle read-back, then repair RustFS expiry processing without deleting evidence manually.
+    recovery: No expiration-eligible diagnostic object remains after the one-hour implementation grace.
   - id: fleet_platform_unhealthy
     severity: page
     query_language: promql

internal/diagnosticstore/reconcile.go

@@ -6,6 +6,7 @@ import (
 	"encoding/xml"
 	"fmt"
 	"net/http"
+	"net/url"
 	"os"
 	"strings"
 	"time"
@@ -28,9 +29,16 @@ type Result struct {
 	HeadroomState        string    `json:"headroom_state"`
 	UsageObservedAt      time.Time `json:"usage_observed_at"`
 	Actions              []string  `json:"actions"`
+	ObjectCount          int       `json:"object_count"`
+	OldestObjectModified time.Time `json:"oldest_object_modified_at,omitempty"`
+	ExpirationEligible   int       `json:"expiration_eligible_objects"`
+	NextExpirationAt     time.Time `json:"next_expiration_at,omitempty"`
 }
 
-type Runner struct{ Requester rustfscache.Requester }
+type Runner struct {
+	Requester rustfscache.Requester
+	Now       func() time.Time
+}
 
 type quotaInfo struct {
 	Quota     int64  `json:"quota"`
@@ -59,6 +67,21 @@ type lifecycleRule struct {
 	} `xml:"Expiration"`
 }
 
+type listBucketResult struct {
+	IsTruncated           bool   `xml:"IsTruncated"`
+	NextContinuationToken string `xml:"NextContinuationToken"`
+	Contents              []struct {
+		LastModified string `xml:"LastModified"`
+	} `xml:"Contents"`
+}
+
+type objectInventory struct {
+	Count              int
+	OldestModified     time.Time
+	ExpirationEligible int
+	NextExpiration     time.Time
+}
+
 func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, error) {
 	if err := config.Validate(); err != nil {
 		return Result{}, err
@@ -71,6 +94,10 @@ func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, err
 		return Result{}, err
 	}
 	defer clear(root.SecretKey)
+	now := time.Now().UTC()
+	if r.Now != nil {
+		now = r.Now().UTC()
+	}
 	state, usage, err := r.inspect(ctx, root, config)
 	if err != nil {
 		return Result{}, err
@@ -82,7 +109,12 @@ func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, err
 		result.CurrentUsageBytes = usage.CurrentUsage
 		result.RemainingQuotaBytes = usage.RemainingQuota
 		result.UsagePercentage = usage.UsagePercentage
-		result.UsageObservedAt = time.Now().UTC()
+		result.UsageObservedAt = now
+		inventory, inventoryErr := r.listObjects(ctx, root, config, now)
+		if inventoryErr != nil {
+			return Result{}, inventoryErr
+		}
+		applyInventory(&result, inventory)
 		result.HeadroomState = "sufficient"
 		if usage.RemainingQuota < config.MinimumHeadroom {
 			result.HeadroomState = "below-minimum"
@@ -111,7 +143,12 @@ func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, err
 	result.CurrentUsageBytes = usageAfter.CurrentUsage
 	result.RemainingQuotaBytes = usageAfter.RemainingQuota
 	result.UsagePercentage = usageAfter.UsagePercentage
-	result.UsageObservedAt = time.Now().UTC()
+	result.UsageObservedAt = now
+	inventory, err := r.listObjects(ctx, root, config, now)
+	if err != nil {
+		return Result{}, err
+	}
+	applyInventory(&result, inventory)
 	if usageAfter.RemainingQuota < config.MinimumHeadroom {
 		result.HeadroomState = "below-minimum"
 	} else {
@@ -120,6 +157,71 @@ func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, err
 	return result, nil
 }
 
+func (r Runner) listObjects(ctx context.Context, root rustfscache.Credential, config Config, now time.Time) (objectInventory, error) {
+	const maximumPages = 1024
+	inventory := objectInventory{}
+	token := ""
+	for page := 0; page < maximumPages; page++ {
+		query := url.Values{"list-type": {"2"}, "prefix": {config.Prefix + "/"}}
+		if token != "" {
+			query.Set("continuation-token", token)
+		}
+		response, err := r.Requester.Do(ctx, root, http.MethodGet, "/"+config.Bucket+"?"+query.Encode(), "", nil)
+		if err != nil {
+			return objectInventory{}, fmt.Errorf("list diagnostic objects: %w", err)
+		}
+		if response.StatusCode != http.StatusOK {
+			return objectInventory{}, responseError("list diagnostic objects", response)
+		}
+		var listed listBucketResult
+		if err := xml.Unmarshal(response.Body, &listed); err != nil {
+			return objectInventory{}, fmt.Errorf("decode diagnostic object listing: %w", err)
+		}
+		for _, object := range listed.Contents {
+			modified, err := time.Parse(time.RFC3339Nano, object.LastModified)
+			if err != nil {
+				return objectInventory{}, fmt.Errorf("decode diagnostic object timestamp: %w", err)
+			}
+			modified = modified.UTC()
+			expires := lifecycleExpirationAt(modified, config.RetentionDays)
+			inventory.Count++
+			if inventory.OldestModified.IsZero() || modified.Before(inventory.OldestModified) {
+				inventory.OldestModified = modified
+			}
+			if !now.Before(expires) {
+				inventory.ExpirationEligible++
+			}
+			if inventory.NextExpiration.IsZero() || expires.Before(inventory.NextExpiration) {
+				inventory.NextExpiration = expires
+			}
+		}
+		if !listed.IsTruncated {
+			return inventory, nil
+		}
+		if listed.NextContinuationToken == "" {
+			return objectInventory{}, fmt.Errorf("truncated diagnostic object listing omitted continuation token")
+		}
+		token = listed.NextContinuationToken
+	}
+	return objectInventory{}, fmt.Errorf("diagnostic object listing exceeded %d pages", maximumPages)
+}
+
+func lifecycleExpirationAt(modified time.Time, retentionDays int) time.Time {
+	candidate := modified.UTC().Add(time.Duration(retentionDays) * 24 * time.Hour)
+	midnight := candidate.Truncate(24 * time.Hour)
+	if candidate.Equal(midnight) {
+		return midnight
+	}
+	return midnight.Add(24 * time.Hour)
+}
+
+func applyInventory(result *Result, inventory objectInventory) {
+	result.ObjectCount = inventory.Count
+	result.OldestObjectModified = inventory.OldestModified
+	result.ExpirationEligible = inventory.ExpirationEligible
+	result.NextExpirationAt = inventory.NextExpiration
+}
+
 func (r Runner) inspect(ctx context.Context, root rustfscache.Credential, config Config) (string, quotaStats, error) {
 	bucket, err := r.Requester.Do(ctx, root, http.MethodHead, "/"+config.Bucket, "", nil)
 	if err != nil {

internal/diagnosticstore/reconcile_test.go

@@ -6,7 +6,9 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/NDDev-OpenNetwork/github-actions/internal/rustfscache"
 )
@@ -16,6 +18,7 @@ type fakeRequester struct {
 	quota     int64
 	usage     int64
 	lifecycle []byte
+	objects   []time.Time
 }
 
 func (f *fakeRequester) Do(_ context.Context, _ rustfscache.Credential, method, path, _ string, body []byte) (rustfscache.Response, error) {
@@ -49,6 +52,14 @@ func (f *fakeRequester) Do(_ context.Context, _ rustfscache.Credential, method,
 	case method == http.MethodPut && path == "/example-diagnostics?lifecycle":
 		f.lifecycle = append([]byte(nil), body...)
 		return rustfscache.Response{StatusCode: http.StatusOK}, nil
+	case method == http.MethodGet && strings.HasPrefix(path, "/example-diagnostics?list-type=2&prefix=diagnostics%2Fv1%2F"):
+		var payload strings.Builder
+		payload.WriteString("<ListBucketResult><IsTruncated>false</IsTruncated>")
+		for _, modified := range f.objects {
+			payload.WriteString("<Contents><LastModified>" + modified.UTC().Format(time.RFC3339Nano) + "</LastModified></Contents>")
+		}
+		payload.WriteString("</ListBucketResult>")
+		return rustfscache.Response{StatusCode: http.StatusOK, Body: []byte(payload.String())}, nil
 	default:
 		return rustfscache.Response{StatusCode: http.StatusBadRequest}, nil
 	}
@@ -113,6 +124,23 @@ func TestReportsLowRemoteHeadroom(t *testing.T) {
 	}
 }
 
+func TestObjectInventoryUsesLifecycleMidnightEligibility(t *testing.T) {
+	config, remote := diagnosticStoreFixture(t)
+	remote.bucket, remote.quota = true, config.QuotaBytes
+	remote.lifecycle = lifecycleDocument(config)
+	modified := time.Date(2026, 8, 19, 0, 36, 58, 0, time.UTC)
+	remote.objects = []time.Time{modified}
+	now := time.Date(2026, 8, 26, 3, 37, 39, 0, time.UTC)
+	result, err := (Runner{Requester: remote, Now: func() time.Time { return now }}).Run(context.Background(), config, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.ObjectCount != 1 || result.OldestObjectModified != modified || result.ExpirationEligible != 0 ||
+		result.NextExpirationAt != time.Date(2026, 8, 27, 0, 0, 0, 0, time.UTC) {
+		t.Fatalf("unexpected lifecycle inventory: %+v", result)
+	}
+}
+
 func TestRejectsQuotaWithoutHeadroom(t *testing.T) {
 	config := Config{SchemaVersion: 1, Endpoint: "https://192.0.2.1:9002", Region: "us-east-1", CAFile: "/tmp/ca",
 		RootAccessKeyFile: "/tmp/access", RootSecretKeyFile: "/tmp/secret", Bucket: "example-diagnostics", Prefix: "diagnostics/v1",

internal/diagnosticstoreobserve/observe.go

@@ -119,6 +119,18 @@ func Render(snapshot Snapshot, now time.Time) string {
 	gauge("gha_diagnostic_storage_usage_percent", "Percentage of the diagnostic bucket hard quota currently used.", snapshot.Result.UsagePercentage)
 	gauge("gha_diagnostic_storage_growth_bytes_per_second", "Non-negative growth rate between the two latest successful signed snapshots.", snapshot.GrowthBytesPerSecond)
 	gauge("gha_diagnostic_storage_forecast_exhaustion_seconds", "Seconds to hard-quota exhaustion at the latest positive growth rate, or -1 when not forecastable.", snapshot.ForecastExhaustionSeconds)
+	gauge("gha_diagnostic_storage_objects", "Objects retained under the diagnostic prefix.", float64(snapshot.Result.ObjectCount))
+	oldestObjectAge := float64(-1)
+	if !snapshot.Result.OldestObjectModified.IsZero() && !snapshot.Result.OldestObjectModified.After(now) {
+		oldestObjectAge = now.Sub(snapshot.Result.OldestObjectModified).Seconds()
+	}
+	gauge("gha_diagnostic_storage_oldest_object_age_seconds", "Age of the oldest retained diagnostic object, or -1 when empty.", oldestObjectAge)
+	gauge("gha_diagnostic_storage_expiration_eligible_objects", "Objects retained after their S3 lifecycle UTC-midnight eligibility boundary.", float64(snapshot.Result.ExpirationEligible))
+	nextExpiration := float64(-1)
+	if !snapshot.Result.NextExpirationAt.IsZero() {
+		nextExpiration = max(0, snapshot.Result.NextExpirationAt.Sub(now).Seconds())
+	}
+	gauge("gha_diagnostic_storage_next_expiration_seconds", "Seconds until the earliest retained object reaches its lifecycle eligibility boundary, or -1 when empty.", nextExpiration)
 	return output.String()
 }
 

internal/diagnosticstoreobserve/observe_test.go

@@ -34,6 +34,26 @@ func TestSampleRendersHeadroomAndForecast(t *testing.T) {
 	}
 }
 
+func TestRetentionMetricsExposeEligibleObjectsAndNextBoundary(t *testing.T) {
+	now := time.Date(2026, 8, 26, 3, 0, 0, 0, time.UTC)
+	snapshot := Snapshot{CapturedAt: now, Result: diagnosticstore.Result{
+		StateAfter: "managed", HeadroomState: "sufficient", ObjectCount: 356,
+		OldestObjectModified: time.Date(2026, 8, 19, 0, 36, 58, 0, time.UTC),
+		ExpirationEligible:   0, NextExpirationAt: time.Date(2026, 8, 27, 0, 0, 0, 0, time.UTC),
+	}}
+	metrics := Render(snapshot, now)
+	for _, wanted := range []string{
+		"gha_diagnostic_storage_objects 356\n",
+		"gha_diagnostic_storage_oldest_object_age_seconds 613382\n",
+		"gha_diagnostic_storage_expiration_eligible_objects 0\n",
+		"gha_diagnostic_storage_next_expiration_seconds 75600\n",
+	} {
+		if !strings.Contains(metrics, wanted) {
+			t.Fatalf("metrics missing %q\n%s", wanted, metrics)
+		}
+	}
+}
+
 func TestHealthFailsClosedForLowHeadroomAndStaleness(t *testing.T) {
 	now := time.Date(2026, 8, 21, 8, 0, 0, 0, time.UTC)
 	state := &State{snapshot: Snapshot{CapturedAt: now, Result: diagnosticstore.Result{StateAfter: "managed", HeadroomState: "below-minimum"}}}

internal/observabilitydashboards/dashboards.go

@@ -130,7 +130,7 @@ func (p Panel) Validate() error {
 	if _, ok := map[string]struct{}{"bytes": {}, "count": {}, "percent": {}, "seconds": {}, "state": {}}[p.Unit]; !ok {
 		return fmt.Errorf("unit is invalid")
 	}
-	if !strings.Contains(p.Query, "gha_fleet_") && !strings.Contains(p.Query, "otelcol_exporter_") {
+	if !strings.Contains(p.Query, "gha_fleet_") && !strings.Contains(p.Query, "gha_diagnostic_storage_") && !strings.Contains(p.Query, "otelcol_exporter_") {
 		return fmt.Errorf("query does not use an owned fleet or Collector metric")
 	}
 	return nil

internal/observabilitydashboards/dashboards_test.go

@@ -21,7 +21,7 @@ func TestPublishedDashboardBundleIsValidAndRenderable(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(bundle.Dashboards) != 6 {
+	if len(bundle.Dashboards) != 7 {
 		t.Fatalf("dashboards=%d", len(bundle.Dashboards))
 	}
 	rendered, err := Render(bundle)

internal/observabilitydashboards/openobserve.go

@@ -7,7 +7,7 @@ import (
 
 const managedDescriptionPrefix = "managed-by:gds;dashboard-contract:v1;"
 
-var metricPattern = regexp.MustCompile(`(?:gha_fleet_|otelcol_exporter_)[a-zA-Z0-9_:]*`)
+var metricPattern = regexp.MustCompile(`(?:gha_fleet_|gha_diagnostic_storage_|otelcol_exporter_)[a-zA-Z0-9_:]*`)
 
 type OpenObserveDashboard struct {
 	Version                 int              `json:"version"`

internal/observabilitydashboards/openobserve_test.go

@@ -14,7 +14,7 @@ func TestRenderOpenObserveV8IsDeterministicAndManaged(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(dashboards) != 6 {
+	if len(dashboards) != 7 {
 		t.Fatalf("dashboards=%d", len(dashboards))
 	}
 	for _, dashboard := range dashboards {
@@ -29,7 +29,7 @@ func TestRenderOpenObserveV8IsDeterministicAndManaged(t *testing.T) {
 				t.Fatalf("invalid panel %#v", panel)
 			}
 			stream := panel.Queries[0].Fields.Stream
-			if !strings.HasPrefix(stream, "gha_fleet_") && !strings.HasPrefix(stream, "otelcol_exporter_") {
+			if !strings.HasPrefix(stream, "gha_fleet_") && !strings.HasPrefix(stream, "gha_diagnostic_storage_") && !strings.HasPrefix(stream, "otelcol_exporter_") {
 				t.Fatalf("panel %q stream is not an owned metric: %q", panel.ID, stream)
 			}
 			if panel.Layout.I != index+1 || panel.Layout.W != 96 || panel.Layout.H != 9 {

internal/observabilityrules/rules_test.go

@@ -10,8 +10,8 @@ func TestRepositoryBundleIsValid(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(bundle.Rules) != 14 {
-		t.Fatalf("rules = %d, want 14", len(bundle.Rules))
+	if len(bundle.Rules) != 15 {
+		t.Fatalf("rules = %d, want 15", len(bundle.Rules))
 	}
 }