fix(agent): set WhenRequired on the transfer manager too, not just the client

Review of #342 found the client-only option never reached the code path that
uploads real snapshots. manager.Uploader carries its OWN
RequestChecksumCalculation (NewUploader hardcodes WhenSupported) and never reads
the s3.Client option, so any snapshot over the manager's 5 MiB part size — every
real etcd snapshot — took the multipart branch and re-stamped the CRC32
STREAMING-UNSIGNED-PAYLOAD-TRAILER that Ceph RGW rejects. The previous fix only
covered the degenerate sub-5-MiB single-part case.

- Derive both knobs from one package-level constant (snapshotChecksumCalculation)
  and set it on the uploader via newSnapshotUploader as well as the client, so the
  two independent settings cannot silently drift.
- Add TestUploadS3StreamMultipartNoChecksumTrailer: drives the real uploadS3Stream
  against a TLS httptest server (trusted via AWS_CA_BUNDLE) with a >5 MiB body and
  asserts no UploadPart carries a checksum trailer/algorithm. Fails without the
  uploader option; a check on s3.Options alone stays green while the bug is live.
- Make TestS3ClientChecksumWhenRequired hermetic (AWS_CONFIG_FILE/… = /dev/null,
  empty AWS_PROFILE) and assert the response-side setting; keep it as a
  client-option unit check only.
- Set ResponseChecksumValidation=WhenRequired symmetrically for the restore
  download path (same mismatch class; defensive, the downloader sets no
  ChecksumMode today).
- Rewrite the agent.go comment that wrongly claimed a plain PutObject, and fix the
  stale "temp file it then multipart-uploads" sentence in docs/concepts.md.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Andrey Kolkov <andrey.kolkov@aenix.io>

Author: Andrey Kolkov

docs/concepts.md

@@ -449,7 +449,7 @@ Two surfaces, one agent. The operator image doubles as a snapshot agent: `main.g
 An `EtcdSnapshot` is a one-shot record. The controller resolves `spec.clusterRef`, builds a Job (owned by the `EtcdSnapshot`, with `ttlSecondsAfterFinished` so it self-GCs, `automountServiceAccountToken: false`, and a restricted security context), and tracks `status.phase` through `Pending → Started → Complete | Failed`. The Job's Pod runs the snapshot agent, which:
 
 1. dials the cluster's client Service — TLS material (server CA + operator-client cert) is mounted from the same Secrets the operator dials with, and when `status.authEnabled` is latched the agent authenticates as `root` using the password from `spec.auth.rootCredentialsSecretRef`;
-2. streams `clientv3` `Maintenance.Snapshot` to a local file (PVC destination) or a temp file it then multipart-uploads to S3, hashing (sha256) and counting bytes as it goes;
+2. streams `clientv3` `Maintenance.Snapshot` to a local file (PVC destination) or directly to S3 via the transfer manager — no local staging, multipart above the manager's 5 MiB part size — hashing (sha256) and counting bytes as it goes;
 3. prints a marker line — `snapshot uploaded: uri="..." size=N sha256=<hex>` — that the controller scans out of the Pod log (via an uncached `APIReader` to find the Pod and the typed Clientset to read its log) to populate `status.artifact{uri,sizeBytes,checksum}`.
 
 Why parse a log line rather than have the agent write status? The agent has no Kubernetes API access by design (`automountServiceAccountToken: false`), so the controller — which does — is the one that records the result. The marker is the agent→controller channel. Terminal phases are sticky: a `Complete`/`Failed` snapshot is a historical record and never re-runs.

internal/agent/agent.go

@@ -148,10 +148,37 @@ func (d destination) localPath(name string) string {
 	return base + "/" + name + ".db"
 }
 
+// snapshotChecksumCalculation is the S3 request-checksum policy the agent applies
+// on the upload path. It has to be set in TWO independent places, because the
+// snapshot upload touches two SDK layers that each carry their own copy of this
+// knob and do NOT share it:
+//
+//   - the s3.Client option (s3Client, below) — covers the single-part PutObject
+//     the transfer manager issues for a sub-5-MiB body, plus HeadObject;
+//   - the transfer manager's own Uploader.RequestChecksumCalculation
+//     (newSnapshotUploader in snapshot.go) — covers the multipart parts the
+//     manager issues for a body over its 5-MiB part size, which is every real
+//     etcd snapshot. manager.NewUploader hardcodes this field to WhenSupported
+//     and never reads the client option, so setting only the client option
+//     leaves the multipart path — the one that matters — broken.
+//
+// The two are separate knobs that MUST agree; deriving both from this single
+// constant is what keeps them from silently drifting (delete either site and the
+// failure returns — for the multipart case, invisibly).
+//
+// Why WhenRequired: since early 2025 aws-sdk-go-v2 defaults to WhenSupported,
+// which stamps a CRC32 on every PutObject/UploadPart — over HTTPS that rides as a
+// STREAMING-UNSIGNED-PAYLOAD-TRAILER. Ceph RGW (the confirmed case) rejects the
+// accompanying header with "InvalidArgument: x-amz-content-sha256 must be
+// UNSIGNED-PAYLOAD, STREAMING-AWS4-HMAC-SHA256-PAYLOAD or a valid sha256 value".
+// WhenRequired attaches a checksum only when the operation mandates one (none for
+// a plain PutObject or UploadPart) and is equally accepted by real AWS S3.
+const snapshotChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
+
 // s3Client builds an S3 client honoring a custom endpoint, region and
 // path-style addressing. Credentials come from AWS_ACCESS_KEY_ID /
 // AWS_SECRET_ACCESS_KEY in the environment (loaded by the default config
-// chain).
+// chain). Used by BOTH the upload (snapshot) and download (restore) paths.
 func (d destination) s3Client(ctx context.Context) (*s3.Client, error) {
 	region := d.s3Region
 	if region == "" {
@@ -166,18 +193,20 @@ func (d destination) s3Client(ctx context.Context) (*s3.Client, error) {
 			o.BaseEndpoint = aws.String(d.s3Endpoint)
 		}
 		o.UsePathStyle = d.s3PathStyle
-		// Only add a request checksum when the operation actually requires one.
-		//
-		// Since early 2025, aws-sdk-go-v2 defaults RequestChecksumCalculation to
-		// WhenSupported, which attaches a CRC32 to every PutObject. Many
-		// S3-compatible backends that are not AWS S3 (Ceph RGW, some MinIO and
-		// Cloudflare R2 versions, etc.) reject the accompanying header with
-		// "InvalidArgument: x-amz-content-sha256 must be UNSIGNED-PAYLOAD,
-		// STREAMING-AWS4-HMAC-SHA256-PAYLOAD or a valid sha256 value", so the
-		// snapshot upload fails against them. WhenRequired keeps uploads working
-		// on those backends and is equally accepted by real AWS S3, which does
-		// not require a checksum on a plain PutObject.
-		o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
+		// Request-checksum policy for this client. This is only ONE of the two
+		// sites that need it — the transfer manager carries its own copy for the
+		// multipart path (see snapshotChecksumCalculation and newSnapshotUploader).
+		// LoadDefaultConfig would otherwise populate this from
+		// AWS_REQUEST_CHECKSUM_CALCULATION, but the agent's environment is a closed
+		// list built by the controller (no operator can set that var on the Pod),
+		// so we pin it here unconditionally rather than leave an escape hatch that
+		// nothing can reach.
+		o.RequestChecksumCalculation = snapshotChecksumCalculation
+		// Symmetric response-side setting for the restore/download path (this
+		// client is shared with downloadS3): don't demand a response checksum the
+		// backend may not emit — the same AWS-default-vs-S3-compatible mismatch
+		// class. Harmless on the upload path, which issues no checksum-bearing GET.
+		o.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
 	}), nil
 }
 

internal/agent/agent_test.go

@@ -36,7 +36,30 @@ func TestObjectKey(t *testing.T) {
 	}
 }
 
+// hermeticAWSEnv isolates the aws-sdk-go-v2 default config chain from the host:
+// LoadDefaultConfig otherwise reads ~/.aws/config and AWS_PROFILE, so a runner
+// whose AWS_PROFILE points at a missing profile would fail these tests for
+// reasons unrelated to the assertion. It also pins static credentials and a
+// region so signing never reaches the EC2/ECS metadata endpoint.
+func hermeticAWSEnv(t *testing.T) {
+	t.Helper()
+	t.Setenv("AWS_CONFIG_FILE", "/dev/null")
+	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", "/dev/null")
+	t.Setenv("AWS_PROFILE", "")
+	t.Setenv("AWS_ACCESS_KEY_ID", "test")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "test")
+	t.Setenv("AWS_REGION", "us-east-1")
+}
+
+// TestS3ClientChecksumWhenRequired is a unit check on the s3.Client OPTION only.
+// It does NOT exercise the path that actually broke: the transfer manager carries
+// its own RequestChecksumCalculation and overrides this on the multipart upload.
+// The end-to-end contract — that a real (>5 MiB) multipart snapshot upload emits
+// no checksum trailer — lives in TestUploadS3StreamMultipartNoChecksumTrailer
+// (snapshot_test.go). Keep both: this pins the client/HeadObject/single-part
+// knob, that one pins multipart.
 func TestS3ClientChecksumWhenRequired(t *testing.T) {
+	hermeticAWSEnv(t)
 	d := destination{
 		kind:        "s3",
 		s3Endpoint:  "https://s3.example.internal",
@@ -48,12 +71,17 @@ func TestS3ClientChecksumWhenRequired(t *testing.T) {
 	}
 	o := c.Options()
 	// Non-AWS S3-compatible backends (Ceph RGW, some MinIO/R2) reject the
-	// aws-sdk-go-v2 default (WhenSupported) flexible checksum on PutObject, so
-	// the agent must only add a checksum when the operation requires one.
+	// aws-sdk-go-v2 default (WhenSupported) flexible checksum, so the agent must
+	// only add a checksum when the operation requires one.
 	if o.RequestChecksumCalculation != aws.RequestChecksumCalculationWhenRequired {
 		t.Errorf("RequestChecksumCalculation = %v, want WhenRequired (%v)",
 			o.RequestChecksumCalculation, aws.RequestChecksumCalculationWhenRequired)
 	}
+	// Symmetric response-side setting covering the restore/download path.
+	if o.ResponseChecksumValidation != aws.ResponseChecksumValidationWhenRequired {
+		t.Errorf("ResponseChecksumValidation = %v, want WhenRequired (%v)",
+			o.ResponseChecksumValidation, aws.ResponseChecksumValidationWhenRequired)
+	}
 	if !o.UsePathStyle {
 		t.Error("UsePathStyle = false, want true")
 	}

internal/agent/restore.go

@@ -270,6 +270,11 @@ func downloadS3(ctx context.Context, src destination, stageDir string) (string,
 		return "", err
 	}
 	defer f.Close()
+	// client is built by s3Client, which pins ResponseChecksumValidation to
+	// WhenRequired — so the downloader doesn't demand a response checksum an
+	// S3-compatible backend may not return (symmetric to the upload-side request
+	// checksum). manager.Downloader has no response-validation knob of its own, so
+	// the client option is the only site for this.
 	downloader := manager.NewDownloader(client)
 	if _, err := downloader.Download(ctx, f, &s3.GetObjectInput{
 		Bucket: aws.String(src.s3Bucket),

internal/agent/snapshot.go

@@ -237,13 +237,27 @@ func uploadS3Stream(ctx context.Context, dest destination, key string, body io.R
 	if uid != "" {
 		metadata = map[string]string{snapshotUIDMetaKey: uid} // stamp ownership so a retry recognizes its own object
 	}
-	return uploadStreamHashed(ctx, manager.NewUploader(client), &s3.PutObjectInput{
+	return uploadStreamHashed(ctx, newSnapshotUploader(client), &s3.PutObjectInput{
 		Bucket:   aws.String(dest.s3Bucket),
 		Key:      aws.String(key),
 		Metadata: metadata,
 	}, body)
 }
 
+// newSnapshotUploader builds the S3 transfer manager for snapshot uploads.
+//
+// manager.Uploader carries its OWN RequestChecksumCalculation (NewUploader
+// hardcodes it to WhenSupported) and never consults the s3.Client option, so the
+// multipart path — taken for every body over the 5-MiB part size, i.e. every real
+// etcd snapshot — would re-stamp the CRC32 trailer that the client option was set
+// to avoid. Pin it here too, from the same constant s3Client uses, so the two
+// knobs cannot drift. See snapshotChecksumCalculation.
+func newSnapshotUploader(client *s3.Client) *manager.Uploader {
+	return manager.NewUploader(client, func(u *manager.Uploader) {
+		u.RequestChecksumCalculation = snapshotChecksumCalculation
+	})
+}
+
 // s3Uploader abstracts manager.Uploader.Upload so uploadStreamHashed is testable
 // without a live S3 endpoint.
 type s3Uploader interface {

internal/agent/snapshot_test.go

@@ -11,15 +11,20 @@ You may obtain a copy of the License at
 package agent
 
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/pem"
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -81,6 +86,106 @@ func TestUploadStreamHashed_UploadError(t *testing.T) {
 	}
 }
 
+// TestUploadS3StreamMultipartNoChecksumTrailer is the guard for the actual
+// failure mode. A real etcd snapshot exceeds the transfer manager's 5 MiB part
+// size, so the upload takes the MULTIPART branch, where manager.Uploader's own
+// RequestChecksumCalculation (NOT the s3.Client option) decides whether a CRC32
+// trailer is stamped on each UploadPart. Without newSnapshotUploader pinning it
+// to WhenRequired, every part rides `x-amz-content-sha256:
+// STREAMING-UNSIGNED-PAYLOAD-TRAILER` + `x-amz-trailer: x-amz-checksum-crc32` —
+// exactly the header Ceph RGW rejects.
+//
+// This drives the production uploadS3Stream against a TLS test server (trusted via
+// AWS_CA_BUNDLE, which LoadDefaultConfig wires into the real client, so s3Client
+// and newSnapshotUploader run exactly as in production) with a >5 MiB body, and
+// asserts no request carries a checksum trailer or algorithm. It fails if the
+// uploader option is dropped; a check on s3.Options alone (see agent_test.go)
+// would stay green while this real path stayed broken.
+func TestUploadS3StreamMultipartNoChecksumTrailer(t *testing.T) {
+	hermeticAWSEnv(t)
+
+	type capturedReq struct{ method, sha, trailer, algo string }
+	var mu sync.Mutex
+	var reqs []capturedReq
+
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		mu.Lock()
+		reqs = append(reqs, capturedReq{
+			method:  r.Method,
+			sha:     r.Header.Get("X-Amz-Content-Sha256"),
+			trailer: r.Header.Get("X-Amz-Trailer"),
+			algo:    r.Header.Get("X-Amz-Sdk-Checksum-Algorithm"),
+		})
+		mu.Unlock()
+
+		q := r.URL.Query()
+		switch {
+		case r.Method == http.MethodHead:
+			// ensureObjectAbsent's HeadObject: report absent so the upload proceeds.
+			w.WriteHeader(http.StatusNotFound)
+		case r.Method == http.MethodPost && q.Has("uploads"):
+			io.WriteString(w, `<?xml version="1.0" encoding="UTF-8"?>`+
+				`<InitiateMultipartUploadResult><Bucket>b</Bucket><Key>k</Key>`+
+				`<UploadId>test-upload-id</UploadId></InitiateMultipartUploadResult>`)
+		case r.Method == http.MethodPut && q.Has("partNumber"):
+			w.Header().Set("ETag", `"etag-`+q.Get("partNumber")+`"`)
+			w.WriteHeader(http.StatusOK)
+		case r.Method == http.MethodPost && q.Get("uploadId") != "":
+			io.WriteString(w, `<?xml version="1.0" encoding="UTF-8"?>`+
+				`<CompleteMultipartUploadResult><Bucket>b</Bucket><Key>k</Key>`+
+				`<ETag>"final-etag"</ETag></CompleteMultipartUploadResult>`)
+		default:
+			w.WriteHeader(http.StatusOK)
+		}
+	}))
+	defer srv.Close()
+
+	// Trust the test server's self-signed cert through the SDK's default config
+	// chain, so uploadS3Stream's own s3Client reaches it — no hand-built client.
+	caFile := filepath.Join(t.TempDir(), "ca.pem")
+	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: srv.Certificate().Raw})
+	if err := os.WriteFile(caFile, certPEM, 0o600); err != nil {
+		t.Fatalf("write CA bundle: %v", err)
+	}
+	t.Setenv("AWS_CA_BUNDLE", caFile)
+
+	const bodySize = 12 << 20 // 12 MiB > 5 MiB part size ⇒ three parts ⇒ multipart branch
+	dest := destination{kind: "s3", s3Endpoint: srv.URL, s3Bucket: "b", s3PathStyle: true}
+
+	size, sum, err := uploadS3Stream(context.Background(), dest, "k", bytes.NewReader(make([]byte, bodySize)), "test-uid")
+	if err != nil {
+		t.Fatalf("uploadS3Stream: %v", err)
+	}
+	if size != bodySize {
+		t.Errorf("streamed size = %d, want %d", size, bodySize)
+	}
+	if sum == "" {
+		t.Error("empty sha256 digest")
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	var sawUploadPart bool
+	for _, rq := range reqs {
+		if rq.method == http.MethodPut {
+			sawUploadPart = true
+		}
+		if rq.sha == "STREAMING-UNSIGNED-PAYLOAD-TRAILER" {
+			t.Errorf("%s carried x-amz-content-sha256=STREAMING-UNSIGNED-PAYLOAD-TRAILER — the checksum trailer Ceph RGW rejects", rq.method)
+		}
+		if rq.trailer != "" {
+			t.Errorf("%s carried x-amz-trailer=%q, want none", rq.method, rq.trailer)
+		}
+		if rq.algo != "" {
+			t.Errorf("%s carried x-amz-sdk-checksum-algorithm=%q, want none", rq.method, rq.algo)
+		}
+	}
+	if !sawUploadPart {
+		t.Fatal("no UploadPart (PUT) request seen — the upload did not take the multipart branch, so this test is not exercising the path it guards")
+	}
+}
+
 type fakeHead struct {
 	out *s3.HeadObjectOutput
 	err error