fix(pam): stop retrying rejected recording chunks

Chunk reconciliation retried every failed upload on a five-minute tick with no
give-up, so a chunk the platform can only ever reject was re-POSTed for the
life of the gateway. One stuck queue produced 22,511 errors in three hours in
production, and the burst also saturated the write rate limit.

Treat 400, 403 and 404 as permanent and drop the chunk; keep 401, 429, 5xx and
network failures retriable. A failed removal is reported rather than assumed,
since such a chunk stays queued and keeps failing.

Author: bernie-g

packages/pam/session/chunk_uploader.go

@@ -9,6 +9,7 @@ import (
 	"crypto/sha256"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -263,6 +264,41 @@ func (cu *ChunkUploader) EncryptAndQueueChunk(
 	return pc, nil
 }
 
+// Reconciliation never gives up, so a chunk that can only ever be rejected retries forever.
+func isPermanentUploadFailure(err error) bool {
+	var apiErr *api.APIError
+	if !errors.As(err, &apiErr) {
+		return false
+	}
+	switch apiErr.StatusCode {
+	case http.StatusBadRequest, http.StatusForbidden, http.StatusNotFound:
+		return true
+	default:
+		return false
+	}
+}
+
+// flushSession has already advanced past it, so keeping the file only feeds the retry loop.
+func (cu *ChunkUploader) dropIfPermanent(sessionID string, pc *pendingChunk, err error) {
+	if !isPermanentUploadFailure(err) {
+		return
+	}
+
+	entry := log.Error().
+		Err(err).
+		Str("sessionId", sessionID).
+		Int("chunkIndex", pc.ChunkIndex).
+		Int("ciphertextBytes", len(pc.Ciphertext))
+
+	if rmErr := os.Remove(chunkPendingFile(sessionID, pc.ChunkIndex)); rmErr != nil {
+		entry.AnErr("removeError", rmErr).
+			Msg("Recording chunk permanently rejected but could not be removed; it stays queued and will keep failing")
+		return
+	}
+
+	entry.Msg("Recording chunk permanently rejected by platform; dropped instead of retrying")
+}
+
 func (cu *ChunkUploader) UploadChunk(sessionID string, pc *pendingChunk) error {
 	secrets := cu.credentialsManager.GetRecordingSecrets(sessionID)
 	if secrets == nil {
@@ -281,6 +317,7 @@ func (cu *ChunkUploader) UploadChunk(sessionID string, pc *pendingChunk) error {
 			},
 		)
 		if err != nil {
+			cu.dropIfPermanent(sessionID, pc, err)
 			return fmt.Errorf("presigned PUT mint failed: %w", err)
 		}
 		if err := s3PutCiphertext(presigned.URL, pc.Ciphertext); err != nil {
@@ -305,6 +342,7 @@ func (cu *ChunkUploader) UploadChunk(sessionID string, pc *pendingChunk) error {
 	}
 
 	if err := api.CallPAMSessionChunkMetadata(cu.httpClient, sessionID, secrets.UploadToken, metadataReq); err != nil {
+		cu.dropIfPermanent(sessionID, pc, err)
 		return fmt.Errorf("chunk metadata POST failed: %w", err)
 	}
 

packages/pam/session/chunk_uploader_http_test.go

@@ -0,0 +1,140 @@
+package session
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	"github.com/Infisical/infisical-merge/packages/config"
+	"github.com/go-resty/resty/v2"
+)
+
+// Uses a real HTTP server so the status code travels the same path it does in production.
+func newChunkUploaderAgainst(t *testing.T, status int, body string) (*ChunkUploader, *int) {
+	t.Helper()
+
+	hits := 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits += 1
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(status)
+		_, _ = w.Write([]byte(body))
+	}))
+	t.Cleanup(srv.Close)
+
+	orig := config.INFISICAL_URL
+	config.INFISICAL_URL = srv.URL + "/api"
+	t.Cleanup(func() { config.INFISICAL_URL = orig })
+
+	cm := newTestCredentialsManager(t)
+	cm.recordingSecrets["sess"] = &PAMRecordingSecrets{
+		SessionKey:     make([]byte, 32),
+		UploadToken:    "fake-token-for-test",
+		StorageBackend: storageBackendPostgres,
+		ProjectId:      "proj",
+		SessionId:      "sess",
+	}
+
+	return NewChunkUploader(resty.New(), cm), &hits
+}
+
+func queuedChunk(t *testing.T, sessionID string) *pendingChunk {
+	t.Helper()
+	pc := &pendingChunk{
+		ChunkIndex:     0,
+		StartElapsedMs: 0,
+		EndElapsedMs:   10,
+		IV:             make([]byte, 12),
+		Ciphertext:     []byte("ciphertext"),
+		Sha256:         make([]byte, 32),
+		StorageBackend: storageBackendPostgres,
+	}
+	if err := writePendingChunk(sessionID, pc); err != nil {
+		t.Fatal(err)
+	}
+	return pc
+}
+
+func TestUploadChunk_RealInvalidUploadTokenIsDropped(t *testing.T) {
+	setupTestDir(t)
+
+	cu, hits := newChunkUploaderAgainst(t, http.StatusBadRequest,
+		`{"reqId":"req-x","statusCode":400,"message":"Invalid upload token","error":"PamUploadTokenHashMismatch"}`)
+	pc := queuedChunk(t, "sess")
+
+	if err := cu.UploadChunk("sess", pc); err == nil {
+		t.Fatal("expected UploadChunk to surface the 400")
+	}
+	if *hits != 1 {
+		t.Errorf("server saw %d requests, want 1", *hits)
+	}
+	if _, err := os.Stat(chunkPendingFile("sess", 0)); !os.IsNotExist(err) {
+		t.Error("a 400 is permanent: the chunk must be dropped, not left to retry forever")
+	}
+}
+
+func TestUploadChunk_RealRateLimitIsRetained(t *testing.T) {
+	setupTestDir(t)
+
+	cu, _ := newChunkUploaderAgainst(t, http.StatusTooManyRequests,
+		`{"reqId":"req-y","statusCode":429,"message":"Rate limit exceeded. Please try again in 14 seconds"}`)
+	pc := queuedChunk(t, "sess")
+
+	if err := cu.UploadChunk("sess", pc); err == nil {
+		t.Fatal("expected UploadChunk to surface the 429")
+	}
+	if _, err := os.Stat(chunkPendingFile("sess", 0)); err != nil {
+		t.Errorf("a 429 is transient: the chunk must stay queued for retry: %v", err)
+	}
+}
+
+func TestUploadChunk_RealServerErrorIsRetained(t *testing.T) {
+	setupTestDir(t)
+
+	cu, _ := newChunkUploaderAgainst(t, http.StatusInternalServerError, `{"statusCode":500,"message":"boom"}`)
+	pc := queuedChunk(t, "sess")
+
+	if err := cu.UploadChunk("sess", pc); err == nil {
+		t.Fatal("expected UploadChunk to surface the 500")
+	}
+	if _, err := os.Stat(chunkPendingFile("sess", 0)); err != nil {
+		t.Errorf("a 500 is transient: the chunk must stay queued for retry: %v", err)
+	}
+}
+
+func TestReconcileSession_DrainsPermanentlyRejectedQueue(t *testing.T) {
+	setupTestDir(t)
+
+	cu, hits := newChunkUploaderAgainst(t, http.StatusBadRequest,
+		`{"statusCode":400,"message":"Invalid upload token","error":"PamUploadTokenHashMismatch"}`)
+
+	const queued = 5
+	for i := 0; i < queued; i += 1 {
+		pc := &pendingChunk{
+			ChunkIndex:     i,
+			IV:             make([]byte, 12),
+			Ciphertext:     []byte("ciphertext"),
+			Sha256:         make([]byte, 32),
+			StorageBackend: storageBackendPostgres,
+		}
+		if err := writePendingChunk("sess", pc); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	cu.ReconcileSession("sess")
+	if *hits != queued {
+		t.Errorf("first pass made %d requests, want %d", *hits, queued)
+	}
+
+	before := *hits
+	cu.ReconcileSession("sess")
+	if *hits != before {
+		t.Errorf("second pass made %d more requests; the queue was not drained", *hits-before)
+	}
+
+	if cu.HasPendingChunks("sess") {
+		t.Error("queue still holds permanently rejected chunks")
+	}
+}

packages/pam/session/chunk_uploader_test.go

@@ -5,11 +5,15 @@ import (
 	"crypto/cipher"
 	"crypto/sha256"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"net/http"
 	"os"
 	"path/filepath"
 	"testing"
 	"time"
+
+	"github.com/Infisical/infisical-merge/packages/api"
 )
 
 func setupTestDir(t *testing.T) {
@@ -346,6 +350,62 @@ func TestDeletePendingChunk(t *testing.T) {
 	}
 }
 
+func TestIsPermanentUploadFailure(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"body too large is retryable after re-chunking", &api.APIError{StatusCode: http.StatusRequestEntityTooLarge}, false},
+		{"bad request", &api.APIError{StatusCode: http.StatusBadRequest}, true},
+		{"forbidden gateway", &api.APIError{StatusCode: http.StatusForbidden}, true},
+		{"session gone", &api.APIError{StatusCode: http.StatusNotFound}, true},
+		{"rate limited", &api.APIError{StatusCode: http.StatusTooManyRequests}, false},
+		{"server error", &api.APIError{StatusCode: http.StatusInternalServerError}, false},
+		{"unauthorized retries after token refresh", &api.APIError{StatusCode: http.StatusUnauthorized}, false},
+		{"network error", errors.New("dial tcp: connection refused"), false},
+		{"wrapped bad request", fmt.Errorf("chunk metadata POST failed: %w", &api.APIError{StatusCode: http.StatusBadRequest}), true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := isPermanentUploadFailure(tc.err); got != tc.want {
+				t.Errorf("isPermanentUploadFailure() = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestDropIfPermanent(t *testing.T) {
+	setupTestDir(t)
+	cu := newTestChunkUploader(t, newTestCredentialsManager(t))
+
+	pc := &pendingChunk{
+		ChunkIndex:     0,
+		IV:             make([]byte, 12),
+		Ciphertext:     []byte("ct"),
+		Sha256:         make([]byte, 32),
+		StorageBackend: "postgres",
+	}
+
+	sid := "drop-permanent"
+	if err := writePendingChunk(sid, pc); err != nil {
+		t.Fatal(err)
+	}
+	cu.dropIfPermanent(sid, pc, fmt.Errorf("wrapped: %w", &api.APIError{StatusCode: http.StatusBadRequest}))
+	if _, err := os.Stat(chunkPendingFile(sid, 0)); !os.IsNotExist(err) {
+		t.Error("permanently rejected chunk should have been dropped from the queue")
+	}
+
+	sid = "keep-transient"
+	if err := writePendingChunk(sid, pc); err != nil {
+		t.Fatal(err)
+	}
+	cu.dropIfPermanent(sid, pc, &api.APIError{StatusCode: http.StatusTooManyRequests})
+	if _, err := os.Stat(chunkPendingFile(sid, 0)); err != nil {
+		t.Errorf("rate-limited chunk must stay queued for retry: %v", err)
+	}
+}
+
 func TestPendingChunkJsonStability(t *testing.T) {
 	pc := &pendingChunk{
 		ChunkIndex:     2,