fix(cloud-backups): checksum Azure uploads server-side (CRC64), at every payload size

cloud-backups/README.md

@@ -241,7 +241,9 @@ Relieves disk pressure from a large, **write-only** audit table (per-entity revi
 Design notes:
 - The rename holds an `ACCESS EXCLUSIVE` lock only for a catalog-only operation (sub-second even on large tables). `--lock-timeout` makes it **fail-safe**: on contention the rotate rolls back untouched and retries next run.
 - The `DROP` is the sole irreversible step and runs **only after a verified upload**. By default, verification is a **byte-integrity** gate: the upload completed (on real AWS, S3 verifies a SHA-256 of every part server-side and refuses on mismatch) **+** a `HeadObject` size match **+** a whole-object SHA-256 written as a `<key>.sha256` sidecar. This does **not** by itself prove the archive is *restorable*.
-- `--verify-restore` upgrades that to a **restorability** check: it re-downloads the object, **decrypts** it, runs **`pg_restore -l`** (proves it's a structurally valid, decryptable dump — catches a truncated/corrupt archive or a wrong/lost age passphrase before the source is dropped), and matches the SHA-256. Costs a full re-download — reserve for first/manual runs. **Recommended on Azure**, whose uploads are not server-side checksum-verified.
+- `--verify-restore` upgrades that to a **restorability** check: it re-downloads the object, **decrypts** it, runs **`pg_restore -l`** (proves it's a structurally valid, decryptable dump — catches a truncated/corrupt archive or a wrong/lost age passphrase before the source is dropped), and matches the SHA-256. Costs a full re-download -- reserve for first/manual runs.
+
+  Uploads are checksum-verified server-side on **both** backends, at every payload size: real AWS S3 gets a SHA-256 (per part on a multipart upload, whole-object on a small one) and Azure gets a CRC64 (per block, or whole-object for a payload below the block size). So on those two backends `--verify-restore` is about proving *restorability* -- a valid, decryptable dump -- not about catching corruption in transit. A self-hosted S3-compatible endpoint (MinIO/Ceph) still *receives* the SDK's default CRC32, but whether it validates one is up to that server, so prefer enabling `--verify-restore` there.
 - `--no-drop` runs everything except the drop (and exits non-zero so it can't be mistaken for a completed rotation). `--drop-pending` is the confirm step: it does **not** rotate — it verifies each already-backed-up leftover against its sidecar (+ `pg_restore -l`) and drops only the ones that verify, so you drop the exact object you inspected without re-dumping.
 - Object names are deterministic per archive (`{prefix}-{archive}.dump[.age]`, where `{archive}` carries a unique UTC timestamp): distinct and never overwritten across rotations, while a retry or recovery of the *same* archive reuses the key instead of accumulating duplicates. Point this at a **separate, permanent-retention bucket**, not the expiring DB-backup bucket.
 - Runs as the same DB role the application uses, so the fresh table is owned by (and writable by) the app.

cloud-backups/internal/storage/azure.go

@@ -1,15 +1,20 @@
 package storage
 
 import (
+	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/streaming"
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob"
 )
 
 const (
@@ -44,12 +49,69 @@ func newAzureProvider(ctx context.Context, cfg *Config) (*azureProvider, error)
 	return &azureProvider{client: client, container: cfg.AzureContainer}, nil
 }
 
+// Have the SDK compute a CRC64 and send it as x-ms-content-crc64; Azure validates it
+// server-side and rejects the payload on mismatch. Without it Azure receives NO checksum
+// at all, so a "successful" upload means only that bytes arrived, not that they arrived
+// intact -- and this tool DROPs the source table once an upload verifies. The S3 path
+// opts into the same protection (see newS3Provider/UploadStream); this brings Azure to
+// parity, so the cheap existence/size gate can be trusted there instead of requiring a
+// full re-download via --verify-restore.
+//
+// Cost: ComputeCRC64 hashes from a buffer rather than in-line, and it is applied per
+// BLOCK, so peak memory stays flat in archive size (a 4.5 GB archive costs the same as a
+// 100 MB one). It is not free though: io.ReadAll's append growth re-buffers each block at
+// roughly 2x, which measured ~+100-200 MiB RSS at BlockSize 10 MiB x Concurrency 3 --
+// well inside the cronjob's 1Gi limit, but above a naive BlockSize*Concurrency estimate.
+func azureValidation() blob.TransferValidationType {
+	return blob.TransferValidationTypeComputeCRC64()
+}
+
+func azureUploadStreamOptions() *azblob.UploadStreamOptions {
+	return &azblob.UploadStreamOptions{
+		BlockSize:               AzureBlockSize,
+		Concurrency:             AzureConcurrency,
+		TransactionalValidation: azureValidation(),
+	}
+}
+
+// UploadStream streams the payload to a block blob, checksummed whatever its size.
+//
+// The size split is NOT an optimization -- it is required for the checksum to exist at
+// all. azblob's UploadStream short-circuits to a single-shot Put Blob whenever the whole
+// payload fit in block 0 and is smaller than BlockSize (blockblob/chunkwriting.go
+// commitBlocks), and that path silently DROPS TransactionalValidation, because
+// UploadStreamOptions.getUploadOptions does not copy it. So without this split every
+// .sha256 sidecar and every archive under 10 MiB would upload with no server-side
+// integrity check whatsoever -- exactly the case where a length-preserving corruption
+// would survive the size gate and authorize an irreversible DROP. blockblob.Upload does
+// honour the option, so undersized payloads go through it directly.
 func (p *azureProvider) UploadStream(ctx context.Context, remotePath string, reader io.Reader) error {
-	opts := &azblob.UploadStreamOptions{
-		BlockSize:   AzureBlockSize,
-		Concurrency: AzureConcurrency,
+	// Buffer at most one block to find out which path applies. A bytes.Buffer (rather
+	// than a fixed AzureBlockSize array) keeps a 65-byte sidecar from allocating 10 MiB.
+	var head bytes.Buffer
+	_, err := io.CopyN(&head, reader, AzureBlockSize)
+	switch {
+	case errors.Is(err, io.EOF):
+		// Fewer than BlockSize bytes available: the whole payload is in hand.
+		return p.uploadSingleBlock(ctx, remotePath, head.Bytes())
+	case err != nil:
+		return fmt.Errorf("reading first block of %q: %w", remotePath, err)
 	}
-	_, err := p.client.UploadStream(ctx, p.container, remotePath, reader, opts)
+	_, err = p.client.UploadStream(ctx, p.container, remotePath,
+		io.MultiReader(bytes.NewReader(head.Bytes()), reader), azureUploadStreamOptions())
+	if err != nil && ctx.Err() != nil {
+		return fmt.Errorf("upload interrupted: %w", err)
+	}
+	return err
+}
+
+// uploadSingleBlock puts a payload smaller than one block in a single request, keeping
+// the CRC64 that UploadStream's own single-shot path would have discarded.
+func (p *azureProvider) uploadSingleBlock(ctx context.Context, remotePath string, data []byte) error {
+	blobClient := p.client.ServiceClient().NewContainerClient(p.container).NewBlockBlobClient(remotePath)
+	_, err := blobClient.Upload(ctx, streaming.NopCloser(bytes.NewReader(data)), &blockblob.UploadOptions{
+		TransactionalValidation: azureValidation(),
+	})
 	if err != nil && ctx.Err() != nil {
 		return fmt.Errorf("upload interrupted: %w", err)
 	}

cloud-backups/internal/storage/azure_test.go

@@ -1,13 +1,106 @@
 package storage
 
 import (
+	"bytes"
+	"context"
 	"errors"
 	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
 	"testing"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
 )
 
+func TestAzureUploadStreamOptions(t *testing.T) {
+	opts := azureUploadStreamOptions()
+	if opts.TransactionalValidation == nil {
+		t.Error("TransactionalValidation is nil: blocks would be staged with no x-ms-content-crc64, so Azure cannot reject a corrupted block")
+	}
+	if opts.BlockSize != AzureBlockSize {
+		t.Errorf("BlockSize = %d, want %d", opts.BlockSize, AzureBlockSize)
+	}
+	if opts.Concurrency != AzureConcurrency {
+		t.Errorf("Concurrency = %d, want %d", opts.Concurrency, AzureConcurrency)
+	}
+}
+
+// Every upload must carry a server-verifiable checksum REGARDLESS of size, because a
+// verified upload is what authorizes an irreversible DROP. This asserts it on the wire
+// rather than on the options struct: setting TransactionalValidation is necessary but NOT
+// sufficient, since azblob's UploadStream silently drops it on the single-shot Put Blob
+// path it takes for payloads below BlockSize. Without the size split in UploadStream this
+// test fails for the small cases while the options-struct test above still passes.
+func TestAzureUploadStreamChecksumsEverySize(t *testing.T) {
+	cases := []struct {
+		name      string
+		size      int
+		wantParts int // staged blocks; 0 => single-shot Put Blob
+	}{
+		{"sha256 sidecar sized", 65, 0},
+		{"just under one block", AzureBlockSize - 1, 0},
+		{"exactly one block", AzureBlockSize, 1},
+		{"spans two blocks", AzureBlockSize + 1024, 2},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			var (
+				mu       sync.Mutex
+				bodies   int
+				unsigned []string
+			)
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				io.Copy(io.Discard, r.Body)
+				comp := r.URL.Query().Get("comp")
+				if comp == "block" || comp == "" { // a staged block, or a single-shot Put Blob
+					mu.Lock()
+					bodies++
+					if r.Header.Get("x-ms-content-crc64") == "" {
+						unsigned = append(unsigned, fmt.Sprintf("comp=%q len=%s", comp, r.Header.Get("Content-Length")))
+					}
+					mu.Unlock()
+				}
+				w.Header().Set("ETag", `"0x1"`)
+				w.Header().Set("Last-Modified", "Wed, 30 Jul 2026 00:00:00 GMT")
+				w.WriteHeader(http.StatusCreated)
+			}))
+			defer srv.Close()
+
+			client, err := azblob.NewClientWithNoCredential(srv.URL, nil)
+			if err != nil {
+				t.Fatalf("client: %v", err)
+			}
+			p := &azureProvider{client: client, container: "archives"}
+			// A pipe, not a file: the real caller streams pg_dump | age, so the source is
+			// not seekable and its length is unknown up front.
+			pr, pw := io.Pipe()
+			go func() {
+				pw.Write(bytes.Repeat([]byte("a"), tc.size))
+				pw.Close()
+			}()
+			if err := p.UploadStream(context.Background(), "audit_archive.dump.age", pr); err != nil {
+				t.Fatalf("UploadStream: %v", err)
+			}
+
+			mu.Lock()
+			defer mu.Unlock()
+			if len(unsigned) > 0 {
+				t.Errorf("%d of %d payload request(s) carried NO x-ms-content-crc64: %v", len(unsigned), bodies, unsigned)
+			}
+			wantBodies := tc.wantParts
+			if wantBodies == 0 {
+				wantBodies = 1 // one single-shot Put Blob
+			}
+			if bodies != wantBodies {
+				t.Errorf("payload requests = %d, want %d", bodies, wantBodies)
+			}
+		})
+	}
+}
+
 func TestMapAzureNotFound(t *testing.T) {
 	cases := []struct {
 		name         string