feat: manage backup retention in Klio instead of Kopia

AGENTS.md

@@ -139,6 +139,7 @@ Klio-only assertions) must live outside `machinery` — e.g. under
   - `operator/pkg/config/server.go` ↔ `core/pkg/config/server.go`
   - `operator/pkg/config/client.go` ↔ `core/pkg/config/client.go`
   - `operator/pkg/config/compression.go` ↔ `core/pkg/config/compression.go`
+  - `operator/pkg/config/retention.go` ↔ `core/pkg/config/retention.go`
 
 - When you change a metric in `core/internal/opentelemetry/catalog.go`
   (rename, add, remove, or change a metric's unit, type, or attributes),
@@ -190,28 +191,31 @@ first.** Explain that it bypasses the Kopia server cache, name the race it can
 introduce, and propose the server-routed alternative. Only proceed if they
 confirm after that warning.
 
-- Client- and sidecar-driven paths (backup upload, delete, retention set,
+- Client- and sidecar-driven paths (backup upload, delete, retention apply,
   restore, list; everything under `core/cmd/*`) already route through the server
   via `MultiConnect`/`ConnectTier1`/`ConnectTier2`. Keep them that way — never
   convert one of these to a direct write.
 - The **only** component that writes directly is the server-side backup consumer
   (`core/internal/consumer/`), and only because it has no server connection for
-  those steps: tier1/tier2 retention apply, tier2 relay/migrate, tier2 policy
-  set, and tier1 unpin. This is a deliberate, contained exception — not a pattern
-  to copy, and one that should be removed in the future.
-- A second, narrower exception: `applyGlobalCompressionPolicy` in
+  those steps: tier1/tier2 retention apply (snapshot deletes), tier2
+  relay/migrate, and tier2 compression policy set. This is a deliberate,
+  contained exception — not a pattern to copy, and one that should be removed
+  in the future.
+- A second, narrower exception: `applyGlobalKopiaPolicies` in
   `core/cmd/server/server.go` sets the repository-wide (global) compression
-  policy with a raw `kopia.Client{ConfigFile: ...}`, before the tier's Kopia
+  policy and disables Kopia's own snapshot retention (Klio applies retention
+  itself) with a raw `kopia.Client{ConfigFile: ...}`, before the tier's Kopia
   server starts. This is safe only because no server is running yet to hold a
   stale cache. Do not reuse this pattern once the server is up.
 - A direct write that **rewrites the manifest of a live backup** MUST be followed
-  by `refreshTier1KopiaServer` / `refreshTier2KopiaServer` so the servers
-  reconcile their caches; skipping the refresh is a bug. The tier1 unpin is the
-  canonical case: `kopia snapshot pin` rewrites the snapshot manifest to a *new*
-  ID and deletes the old one, so without a refresh the server keeps serving the
-  now-deleted ID for a backup that still exists, and a later client
-  `klio backup delete` asks Kopia to delete an ID that no longer matches
-  anything: the command fails and the real backup (and its WALs) stay pinned.
+  by `refreshTier2KopiaServer` (or an equivalent tier1 refresh, should such a
+  write come back) so the servers reconcile their caches; skipping the refresh
+  is a bug. The former tier1 unpin was the canonical case: `kopia snapshot pin`
+  rewrites the snapshot manifest to a *new* ID and deletes the old one, so
+  without a refresh the server kept serving the now-deleted ID for a backup
+  that still existed, and a later client `klio backup delete` asked Kopia to
+  delete an ID that no longer matched anything: the command failed and the
+  real backup (and its WALs) stayed pinned. Klio no longer pins snapshots.
 - A direct write that only **deletes** snapshots (the tier1/tier2 retention
   apply) does **not** need a refresh: it removes IDs the server may still list,
   but it never rewrites a live backup's ID, and WAL retention is recomputed from
@@ -226,10 +230,10 @@ refresh of the affected tier.
 
 ### Snapshot identity: manifest ID vs root object ID
 
-A snapshot's **manifest ID is not a stable identity**. `kopia snapshot pin`
-(the tier1 unpin above) rewrites a snapshot's manifest under a new ID and
-deletes the old one, so any code that lists snapshots and then acts on them a
-moment later can be holding an ID that no longer exists. Pick the identity by
+A snapshot's **manifest ID is not a stable identity**. Some Kopia writes
+(`kopia snapshot pin`, for one) rewrite a snapshot's manifest under a new ID
+and delete the old one, so any code that lists snapshots and then acts on them
+a moment later can be holding an ID that no longer exists. Pick the identity by
 what the operation does:
 
 - **Reads that must survive a concurrent rewrite** use the root object ID
@@ -244,13 +248,10 @@ what the operation does:
   so deleting one backup by root ID can take another backup's snapshot with it.
   Delete by manifest ID, and on failure re-list and retry so a concurrent
   rewrite is picked up (`DeleteBackup` in the same package).
-- **The tier1 unpin is a write, not a read, and knowingly accepts the same
-  collision as delete.** The consumer's `getPinnedSnapshots`/`maintainTier2`
-  (`core/internal/consumer/backup.go`) also targets the root object ID, so a
-  root shared with another backup gets unpinned too. This is tolerated only
-  because the step is best-effort and the affected snapshot would be unpinned
-  anyway on the next tier2 migration — it is not a safe pattern to copy for
-  anything that isn't equally tolerant of that collision.
+- **Writes that target the root object ID hit the same collision as delete.**
+  The former tier1 unpin did this knowingly, tolerated only because the step
+  was best-effort. Any new write keyed by root object ID needs the same
+  analysis and must be equally tolerant of acting on another backup's snapshot.
 
 ### Dagger caching issues
 

core/cmd/backup/run.go

@@ -119,20 +119,21 @@ func runBackup(cmd *cobra.Command, _ []string) error {
 
 	backupName, _ := cmd.Flags().GetString("name")
 	opts.Name = backupName
+	opts.SendToTier2 = tier2
 
 	if err := backupExecutor.Start(cmd.Context(), opts); err != nil {
 		return cli.NewCodedError(
 			fmt.Errorf("while starting the backup: %w", err),
 			backupfailure.RepositoryError.ExitCode)
 	}
 
-	if err := backupExecutor.Upload(cmd.Context(), tier2); err != nil {
+	if err := backupExecutor.Upload(cmd.Context()); err != nil {
 		return cli.NewCodedError(
 			fmt.Errorf("while uploading data: %w", err),
 			backupfailure.RepositoryError.ExitCode)
 	}
 
-	metadata, err := backupExecutor.Close(cmd.Context(), tier2)
+	metadata, err := backupExecutor.Close(cmd.Context())
 	if err != nil {
 		return cli.NewCodedError(
 			fmt.Errorf("while closing the backup: %w", err),
@@ -146,6 +147,16 @@ func runBackup(cmd *cobra.Command, _ []string) error {
 			backupfailure.RepositoryError.ExitCode)
 	}
 
+	tier1RetentionPolicy, err := configuration.Tier1RetentionPolicy.MarshalWire()
+	if err != nil {
+		contextLogger.Error(err, "Error while serializing the tier1 retention policy, skipping")
+	}
+
+	tier2RetentionPolicy, err := configuration.Tier2RetentionPolicy.MarshalWire()
+	if err != nil {
+		contextLogger.Error(err, "Error while serializing the tier2 retention policy, skipping")
+	}
+
 	for {
 		//nolint:gosec // postgres timeline is uint32 in practice, fits int32
 		timeline := int32(metadata.Timeline)
@@ -158,7 +169,8 @@ func runBackup(cmd *cobra.Command, _ []string) error {
 			EndWal:                 metadata.EndWAL,
 			SegmentSize:            metadata.SegmentSize,
 			SendToTier2:            tier2,
-			Tier2RetentionPolicy:   marshalTier2RetentionPolicy(cmd.Context(), &configuration),
+			Tier1RetentionPolicy:   tier1RetentionPolicy,
+			Tier2RetentionPolicy:   tier2RetentionPolicy,
 			Tier2CompressionPolicy: marshalTier2CompressionPolicy(cmd.Context(), &configuration),
 		})
 		if err != nil {
@@ -226,33 +238,6 @@ func toKopiaCompressionPolicy(p *config.CompressionPolicy) kopiaWrapper.Compress
 	}
 }
 
-// marshalTier2RetentionPolicy serializes the tier2 retention policy to the
-// JSON representation expected by the WAL server. It returns an empty string
-// when no policy is configured or serialization fails.
-func marshalTier2RetentionPolicy(ctx context.Context, configuration *config.Data) string {
-	if configuration.Tier2RetentionPolicy == nil {
-		return ""
-	}
-
-	policy := kopiaWrapper.RetentionPolicy{
-		KeepLatest:  configuration.Tier2RetentionPolicy.KeepLatest,
-		KeepHourly:  configuration.Tier2RetentionPolicy.KeepHourly,
-		KeepDaily:   configuration.Tier2RetentionPolicy.KeepDaily,
-		KeepWeekly:  configuration.Tier2RetentionPolicy.KeepWeekly,
-		KeepMonthly: configuration.Tier2RetentionPolicy.KeepMonthly,
-		KeepAnnual:  configuration.Tier2RetentionPolicy.KeepAnnual,
-	}
-
-	content, err := json.Marshal(policy)
-	if err != nil {
-		log.FromContext(ctx).Error(err, "Error while serializing the tier2 retention policy, skipping")
-
-		return ""
-	}
-
-	return string(content)
-}
-
 // marshalTier2CompressionPolicy serializes the tier2 compression policy to the
 // JSON representation expected by the WAL server. It is always serialized,
 // even when unconfigured, so that removing the compression section resets the

core/cmd/retention/apply.go

@@ -0,0 +1,105 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package retention
+
+import (
+	"fmt"
+
+	"github.com/cloudnative-pg/machinery/pkg/log"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+
+	"github.com/cloudnative-pg/klio/core/internal/backupfailure"
+	"github.com/cloudnative-pg/klio/core/internal/cli"
+	"github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient"
+	"github.com/cloudnative-pg/klio/core/internal/grpc"
+	"github.com/cloudnative-pg/klio/core/pkg/config"
+)
+
+// applyCmd represents the `retention apply` command.
+//
+//nolint:gochecknoglobals
+var applyCmd = &cobra.Command{
+	Use:   "apply",
+	Short: "Apply the configured retention policy immediately",
+	Long: "Apply the retention policy from the configuration to the target cluster " +
+		"without waiting for the next backup, to free space on demand.",
+	RunE: cli.RunEWithExitCode(runApply),
+}
+
+func runApply(cmd *cobra.Command, _ []string) error {
+	contextLogger := log.FromContext(cmd.Context())
+
+	var configuration config.Data
+
+	// IMPORTANT: this requires this program to be built with "-tags viper_bind_struct"
+	// when using environment variables
+	if err := viper.Unmarshal(&configuration); err != nil {
+		return fmt.Errorf("could not unmarshal configuration: %w", err)
+	}
+
+	// Sets the default values, to be overridden by the user configuration.
+	configuration.SetDefaults()
+
+	if configuration.Client == (config.ClientConfig{}) {
+		return cli.ErrClientSectionIsRequired
+	}
+	if configuration.Client.Wal == (config.WalRepositoryClientConfig{}) {
+		return cli.ErrKlioClientSectionIsRequired
+	}
+
+	if err := configuration.Validate(); err != nil {
+		return fmt.Errorf("configuration validation error: %w", err)
+	}
+
+	tier1RetentionPolicy, err := configuration.Tier1RetentionPolicy.MarshalWire()
+	if err != nil {
+		return fmt.Errorf("while serializing the tier1 retention policy: %w", err)
+	}
+
+	tier2RetentionPolicy, err := configuration.Tier2RetentionPolicy.MarshalWire()
+	if err != nil {
+		return fmt.Errorf("while serializing the tier2 retention policy: %w", err)
+	}
+
+	grpcClient, err := grpcclient.Connect(&configuration.Client, configuration.Client.Wal.Address)
+	if err != nil {
+		return cli.NewCodedError(
+			fmt.Errorf("while connecting to the Klio server: %w", err),
+			backupfailure.RepositoryError.ExitCode)
+	}
+
+	result, err := grpcClient.ApplyRetention(cmd.Context(), &grpc.ApplyRetentionRequest{
+		ClusterName:          configuration.Client.ClusterName,
+		Tier1RetentionPolicy: tier1RetentionPolicy,
+		Tier2RetentionPolicy: tier2RetentionPolicy,
+	})
+	if err != nil {
+		return cli.NewCodedError(
+			fmt.Errorf("while applying retention: %w", err),
+			backupfailure.RepositoryError.ExitCode)
+	}
+
+	if result.GetScheduled() {
+		contextLogger.Info("Retention apply scheduled", "cluster", configuration.Client.ClusterName)
+	}
+
+	return nil
+}