fix(core): storage time ordering, watch state machine, replay safety, API pagination

Storage: fixed-width millisecond timestamp format with a one-time migration
so lexicographic comparisons stay chronological; snapshot auto-IDs get
nanosecond precision; Close is synchronized; commit failures roll back; DSN
paths are URI-escaped; the dedup index is no longer rebuilt every startup;
the in-memory store now dedups by event_id and honors Limit/Offset/SinceSeq.

Event/discovery: stream IDs URL-escape the selector so slash-containing label
selectors round-trip and cannot collide; alias resolution fills empty fields
only and no longer clobbers explicit group/version/kind; rc maps to
replicationcontrollers and rs to replicasets.

Watch: the label selector is now actually applied to List and Watch; the 410
relist path is backoff-throttled, backoff resets after a healthy watch, an
empty list resourceVersion backs off instead of looping forever, non-410
watch errors reconnect without writing a spurious permanent gap, an idle
watch timer forces reconnects, and Run cancels sibling streams and recovers
from panics. Synthetic relist events carry a distinct event_id so unchanged
objects re-listed after a gap survive dedup.

Materialize: a baseline now resets object state (objects deleted during a gap
disappear) and heals open gaps; diff ignores server metadata by path, not by
key name, so user fields named status/uid are kept; nested add/remove carry
Added/Removed flags; dotted paths are escaped; Diff validates the window and
treats a zero before as empty; Snapshot records per-stream watermarks and
writes a TypeSnapshot journal record.

Replay: apply requires a successful dry run; gvrFor covers every policy kind
so Include* toggles work; unsupported kinds are reported as skipped instead
of silently dropped; NodePort services lose clusterIP and nodePort while
headless keeps clusterIP: None; targetNS is always honored; namespace mapping
collisions are detected; plan IDs and field managers are collision-safe and
plans are mutex-guarded against concurrent dry-runs/applies.

API: cursor pagination pushes ingest_seq into the SQL filter (export --server
no longer truncates); gap records expose their payload; /v1/diff requires a
cluster_id; dry-run/apply no longer accept a client-supplied kubeconfig;
plans carry target_context; HTTP server gets timeouts.

Metrics: the watch collector and replay planner now bump the registered
counters, and the server refreshes store-derived gauges on a ticker.

CLI: replay apply works locally and refuses to apply after a failed dry run;
coverage --server follows cursors and shows gap details; timeline
auto-detects namespaces and resolves cluster-scoped objects; diff validates
the window; export writes 0600 files; PVCs are no longer treated as
cluster-scoped; krply-server reports the build version.

cmd/krply-server/main.go

@@ -19,9 +19,10 @@ import (
 	"github.com/krply/krply/internal/metrics"
 	"github.com/krply/krply/internal/replay"
 	"github.com/krply/krply/internal/storage"
+	"github.com/krply/krply/internal/version"
 )
 
-const version = "0.1.0"
+var buildVersion = version.Version
 
 func main() {
 	if err := run(); err != nil {
@@ -86,7 +87,7 @@ func run() error {
 	flag.Parse()
 
 	if *showVer {
-		fmt.Println(version)
+		fmt.Println(buildVersion)
 		return nil
 	}
 
@@ -123,17 +124,40 @@ func run() error {
 
 	m := metrics.New()
 	m.RefreshFromStore(ctx, store)
+	planner.SetMetrics(m)
 
-	srv, err := api.NewServer(store, mat, planner, m, version)
+	// Store-derived gauges are refreshed on a ticker; nothing else in the
+	// process would update degraded streams, gap counts, or store size.
+	go func() {
+		ticker := time.NewTicker(30 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				m.RefreshFromStore(ctx, store)
+			}
+		}
+	}()
+
+	srv, err := api.NewServer(store, mat, planner, m, buildVersion)
 	if err != nil {
 		return fmt.Errorf("build api server: %w", err)
 	}
 
-	httpSrv := &http.Server{Addr: listenAddr, Handler: srv.Handler()}
+	httpSrv := &http.Server{
+		Addr:              listenAddr,
+		Handler:           srv.Handler(),
+		ReadHeaderTimeout: 10 * time.Second,
+		ReadTimeout:       60 * time.Second,
+		WriteTimeout:      2 * time.Minute,
+		IdleTimeout:       2 * time.Minute,
+	}
 
 	errCh := make(chan error, 1)
 	go func() {
-		slog.Info("krply-server listening", "addr", listenAddr, "store", *storePath, "version", version)
+		slog.Info("krply-server listening", "addr", listenAddr, "store", *storePath, "version", buildVersion)
 		errCh <- httpSrv.ListenAndServe()
 	}()
 

cmd/krply/coverage.go

@@ -115,18 +115,26 @@ func coverageServer(ctx context.Context, streamID string) error {
 	q := url.Values{}
 	q.Set("stream_id", streamID)
 	q.Set("record_type", string(event.TypeGap))
-	var page queryv1.EventPage
-	if err := c.get(withQuery("/v1/events", q), &page); err != nil {
-		return err
+	var items []queryv1.Event
+	for {
+		var page queryv1.EventPage
+		if err := c.get(withQuery("/v1/events", q), &page); err != nil {
+			return err
+		}
+		items = append(items, page.Items...)
+		if !page.HasMore || page.NextCursor == "" {
+			break
+		}
+		q.Set("cursor", page.NextCursor)
 	}
-	if len(page.Items) == 0 {
+	if len(items) == 0 {
 		out("no gaps recorded\n")
 		return nil
 	}
-	out("GAPS (%d):\n", len(page.Items))
+	out("GAPS (%d):\n", len(items))
 	w := tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0)
 	fmt.Fprintln(w, "FROM-RV\tTO-RV\tREASON\tDETECTED")
-	for _, it := range page.Items {
+	for _, it := range items {
 		from, to, reason := gapFromEvent(it)
 		fmt.Fprintf(w, "%s\t%s\t%s\t%s\n", from, to, reason, it.ObservedAt.Format(time.RFC3339))
 	}

cmd/krply/diff.go

@@ -51,6 +51,9 @@ func runDiff(cmd *cobra.Command, args []string) error {
 	if since.IsZero() || until.IsZero() {
 		return errors.New("diff requires both --since and --until")
 	}
+	if since.After(until) {
+		return errors.New("diff: --since must not be after --until")
+	}
 
 	if serverURL != "" {
 		return diffServer(cmd.Context(), since, until)

cmd/krply/export.go

@@ -57,7 +57,9 @@ func runExport(cmd *cobra.Command, args []string) error {
 
 	var w io.Writer = os.Stdout
 	if exportOut != "" {
-		fh, err := os.Create(exportOut)
+		// Journal exports can contain recorded ConfigMaps/Secrets, so the file
+		// is created with owner-only permissions.
+		fh, err := os.OpenFile(exportOut, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
 		if err != nil {
 			return err
 		}

cmd/krply/record.go

@@ -120,7 +120,6 @@ var clusterScopedKinds = map[string]bool{
 	"ValidatingWebhookConfiguration": true,
 	"PriorityClass":                  true,
 	"Node":                           true,
-	"PersistentVolumeClaim":          true,
 }
 
 // applyRecordNamespace scopes each namespaced resource to the requested

cmd/krply/replay.go

@@ -41,7 +41,7 @@ var replayPlanCmd = &cobra.Command{
 
 var replayApplyCmd = &cobra.Command{
 	Use:   "apply",
-	Short: "dry-run and apply a replay plan (requires --server)",
+	Short: "dry-run and apply a replay plan (local, or via --server)",
 	Args:  cobra.NoArgs,
 	RunE:  runReplayApply,
 }
@@ -59,8 +59,12 @@ func init() {
 	f.BoolVar(&replayAllowGaps, "allow-gaps", false, "allow a plan with incomplete coverage")
 
 	g := replayApplyCmd.Flags()
-	g.StringVar(&replayPlanID, "plan-id", "", "plan id to apply")
+	g.StringVar(&replayPlanID, "plan-id", "", "plan id to apply (server mode)")
+	g.StringVar(&replaySnapshotID, "snapshot", "", "snapshot id to re-plan and apply (local mode)")
+	g.StringVar(&replaySourceNS, "source-namespace", "", "source namespace to replay (empty = all)")
+	g.StringVar(&replayTargetNS, "target-namespace", "", "target namespace to apply into")
 	g.StringVar(&replayTargetContext, "target-context", "", "target kubeconfig context")
+	g.BoolVar(&replayAllowGaps, "allow-gaps", false, "allow a plan with incomplete coverage")
 	g.BoolVar(&replayConfirm, "confirm", false, "confirm the apply")
 }
 
@@ -101,16 +105,73 @@ func runReplayPlan(cmd *cobra.Command, args []string) error {
 }
 
 func runReplayApply(cmd *cobra.Command, args []string) error {
-	if replayPlanID == "" {
-		return errors.New("replay apply requires --plan-id")
-	}
 	if !replayConfirm {
 		return errors.New("refusing apply without --confirm")
 	}
-	if serverURL == "" {
-		return errors.New("replay apply requires --server pointing at krply-server")
+	if serverURL != "" {
+		if replayPlanID == "" {
+			return errors.New("replay apply --server requires --plan-id")
+		}
+		return replayApplyServer(cmd.Context())
+	}
+	return replayApplyLocal(cmd)
+}
+
+// replayApplyLocal replans the snapshot from the local journal, dry-runs it,
+// and applies it directly. This makes the natural plan-then-apply flow work
+// without a server.
+func replayApplyLocal(cmd *cobra.Command) error {
+	if replaySnapshotID == "" {
+		return errors.New("replay apply (local) requires --snapshot")
+	}
+	store, err := openStore(storePath)
+	if err != nil {
+		return err
+	}
+	defer closeStore(store)
+
+	clusters, err := store.ListClusters(cmd.Context())
+	if err != nil {
+		return err
+	}
+	clusterID, err := firstClusterID(clusters)
+	if err != nil {
+		return err
 	}
 
+	mat := materialize.NewMaterializer(store)
+	policy := replay.DefaultPolicy()
+	policy.AllowGaps = replayAllowGaps
+	planner := replay.NewPlanner(store, mat, policy)
+	plan, err := planner.Plan(cmd.Context(), clusterID, replaySnapshotID, replaySourceNS, replayTargetNS)
+	if err != nil {
+		return err
+	}
+	printPlan(plan)
+
+	dry, err := plan.DryRun(cmd.Context(), kubeconfig, replayTargetContext)
+	if err != nil {
+		return err
+	}
+	printDryRunResult(dry)
+	if !dry.OK {
+		return errors.New("refusing apply: dry run did not pass")
+	}
+	res, err := plan.Apply(cmd.Context(), kubeconfig, replayTargetContext, true)
+	if err != nil {
+		return err
+	}
+	out("applied %d objects\n", res.Applied)
+	for _, e := range res.Errors {
+		warn("apply error: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message)
+	}
+	for _, e := range res.Skipped {
+		warn("skipped: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message)
+	}
+	return nil
+}
+
+func replayApplyServer(ctx context.Context) error {
 	c := newAPIClient(serverURL)
 
 	var dry queryv1.DryRunResult
@@ -120,13 +181,19 @@ func runReplayApply(cmd *cobra.Command, args []string) error {
 	}, &dry); err != nil {
 		return err
 	}
-	out("dry run: applied=%d conflicts=%d errors=%d ok=%v\n", dry.Applied, len(dry.Conflicts), len(dry.Errors), dry.OK)
+	out("dry run: applied=%d conflicts=%d errors=%d skipped=%d ok=%v\n", dry.Applied, len(dry.Conflicts), len(dry.Errors), len(dry.Skipped), dry.OK)
 	for _, ci := range dry.Conflicts {
 		warn("conflict: %s/%s (%s): %s\n", ci.Namespace, ci.Name, ci.Kind, ci.Message)
 	}
 	for _, e := range dry.Errors {
 		warn("dry-run error: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message)
 	}
+	for _, e := range dry.Skipped {
+		warn("dry-run skipped: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message)
+	}
+	if !dry.OK {
+		return errors.New("refusing apply: dry run did not pass")
+	}
 
 	var run queryv1.ReplayRun
 	if err := c.post("/v1/replay-runs", map[string]any{
@@ -140,12 +207,28 @@ func runReplayApply(cmd *cobra.Command, args []string) error {
 	for _, e := range run.Errors {
 		warn("apply error: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message)
 	}
+	for _, e := range run.Skipped {
+		warn("skipped: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message)
+	}
 	if run.Status != "" {
 		out("status: %s\n", run.Status)
 	}
 	return nil
 }
 
+func printDryRunResult(d *replay.DryRunResult) {
+	out("dry run: applied=%d conflicts=%d errors=%d skipped=%d ok=%v\n", d.Applied, len(d.Conflicts), len(d.Errors), len(d.Skipped), d.OK)
+	for _, ci := range d.Conflicts {
+		warn("conflict: %s/%s (%s): %s\n", ci.Namespace, ci.Name, ci.Kind, ci.Message)
+	}
+	for _, e := range d.Errors {
+		warn("dry-run error: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message)
+	}
+	for _, e := range d.Skipped {
+		warn("dry-run skipped: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message)
+	}
+}
+
 func replayPlanServer(ctx context.Context) error {
 	c := newAPIClient(serverURL)
 	var plan queryv1.ReplayPlan

cmd/krply/timeline.go

@@ -102,10 +102,32 @@ func timelineServer(ctx context.Context, name string, since time.Time) error {
 	if err := c.get("/v1/streams", &streams); err != nil {
 		return err
 	}
+
+	// Auto-detect the namespace from the journal when it was not provided, so
+	// namespaced objects work without --namespace.
+	namespace := timelineNamespace
+	if namespace == "" {
+		q := url.Values{}
+		q.Set("name", name)
+		if timelineKind != "" {
+			q.Set("kind", timelineKind)
+		}
+		var page queryv1.EventPage
+		if err := c.get(withQuery("/v1/events", q), &page); err != nil {
+			return err
+		}
+		for _, it := range page.Items {
+			if it.Namespace != "" {
+				namespace = it.Namespace
+				break
+			}
+		}
+	}
+
 	var match *queryv1.Stream
 	for i := range streams {
 		s := &streams[i]
-		if timelineNamespace != "" && s.Namespace != "" && s.Namespace != timelineNamespace {
+		if namespace != "" && s.Namespace != "" && s.Namespace != namespace {
 			continue
 		}
 		if timelineKind != "" && !strings.EqualFold(s.Kind, timelineKind) {
@@ -115,13 +137,13 @@ func timelineServer(ctx context.Context, name string, since time.Time) error {
 		break
 	}
 	if match == nil {
-		return fmt.Errorf("no stream found for object %s/%s", nsLabel(timelineNamespace), name)
+		return fmt.Errorf("no stream found for object %s/%s", nsLabel(namespace), name)
 	}
 
 	ref := api.EncodeObjectRef(storage.ObjectRef{
 		ClusterID: match.ClusterID,
 		StreamID:  match.ID,
-		Namespace: timelineNamespace,
+		Namespace: namespace,
 		Name:      name,
 	})
 	path := "/v1/objects/" + url.PathEscape(ref) + "/history"
@@ -191,7 +213,8 @@ func streamsForObject(ctx context.Context, store storage.Store, namespace, kind
 }
 
 // resolveObjectNamespace finds the namespace of an object when it was not
-// provided, by searching the journal for the object's name.
+// provided, by searching the journal for the object's name. Cluster-scoped
+// objects (empty namespace everywhere) resolve to "".
 func resolveObjectNamespace(ctx context.Context, store storage.Store, name, namespace, kind string) (string, error) {
 	if namespace != "" {
 		return namespace, nil
@@ -205,7 +228,11 @@ func resolveObjectNamespace(ctx context.Context, store storage.Store, name, name
 			return recs[i].Resource.Namespace, nil
 		}
 	}
-	return "", fmt.Errorf("could not determine namespace for object %q (pass --namespace)", name)
+	if len(recs) == 0 {
+		return "", fmt.Errorf("no records found for object %q", name)
+	}
+	// Every matching record is cluster-scoped.
+	return "", nil
 }
 
 type tlRow struct {