riverlog: cap total persisted log size

Snoozed jobs can run repeatedly without consuming attempts, so river:log
entries may grow without bound over long-running jobs. Large metadata
payloads then amplify memory use on each future append.

Add a new `MiddlewareConfig` field `MaxTotalBytes` to bound the total
serialized `river:log` payload. When the new attempt is appended and the
result exceeds the cap, oldest entries are dropped first while always
retaining the newest entry.

The middleware now defaults `MaxTotalBytes` to 8 MB. New tests verify
that oldest entries are pruned when the cap is reached, a tiny cap still
retains the newest entry, and the default is applied.

Author: bgentry

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- `riverlog.Middleware` now supports `MiddlewareConfig.MaxTotalBytes` (default 8 MB) to cap total persisted `river:log` history per job. When the cap is exceeded, oldest log entries are dropped first while retaining the newest entry. [PR #1157](https://github.com/riverqueue/river/pull/1157).
+
 ## [0.31.0] - 2026-02-21
 
 ### Added

riverlog/river_log.go

@@ -17,9 +17,11 @@ import (
 )
 
 const (
-	maxSizeMB    = 2
-	maxSizeBytes = maxSizeMB * 1024 * 1024
-	metadataKey  = "river:log"
+	maxSizeMB      = 2
+	maxSizeBytes   = maxSizeMB * 1024 * 1024
+	maxTotalSizeMB = 8
+	maxTotalBytes  = maxTotalSizeMB * 1024 * 1024
+	metadataKey    = "river:log"
 )
 
 type contextKey struct{}
@@ -77,6 +79,16 @@ type MiddlewareConfig struct {
 	//
 	// Defaults to 2 MB (which is per job attempt).
 	MaxSizeBytes int
+
+	// MaxTotalBytes is the maximum total size of all persisted river logs for a
+	// job attempt history. If appending the latest attempt would exceed this
+	// size, oldest log entries are dropped first.
+	//
+	// The latest entry is always retained, even if doing so means the resulting
+	// payload exceeds MaxTotalBytes.
+	//
+	// Defaults to 8 MB.
+	MaxTotalBytes int
 }
 
 // NewMiddleware initializes a new Middleware with the given slog handler
@@ -136,6 +148,7 @@ func defaultConfig(config *MiddlewareConfig) *MiddlewareConfig {
 	}
 
 	config.MaxSizeBytes = cmp.Or(config.MaxSizeBytes, maxSizeBytes)
+	config.MaxTotalBytes = cmp.Or(config.MaxTotalBytes, maxTotalBytes)
 
 	return config
 }
@@ -193,18 +206,49 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu
 			logData = logData[0:m.config.MaxSizeBytes]
 		}
 
-		allLogDataBytes, err := json.Marshal(append(existingLogData.RiverLog, logAttempt{
+		allLogDataBytes, numDroppedEntries, err := marshalLogDataWithCap(append(existingLogData.RiverLog, logAttempt{
 			Attempt: job.Attempt,
 			Log:     logData,
-		}))
+		}), m.config.MaxTotalBytes)
 		if err != nil {
 			m.Logger.ErrorContext(ctx, m.Name+": Error marshaling log data",
 				slog.Any("error", err),
 			)
 		}
 
+		if numDroppedEntries > 0 {
+			m.Logger.WarnContext(ctx, m.Name+": Logs size exceeded total maximum; dropping oldest entries",
+				slog.Int("max_total_size", m.config.MaxTotalBytes),
+				slog.Int("num_entries_dropped", numDroppedEntries),
+			)
+		}
+
 		metadataUpdates[metadataKey] = json.RawMessage(allLogDataBytes)
 	}()
 
 	return doInner(ctx)
 }
+
+func marshalLogDataWithCap(allLogData []logAttempt, maxTotalBytes int) ([]byte, int, error) {
+	allLogDataBytes, err := json.Marshal(allLogData)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	if maxTotalBytes <= 0 || len(allLogDataBytes) <= maxTotalBytes {
+		return allLogDataBytes, 0, nil
+	}
+
+	// Drop oldest entries first, while always retaining the latest one.
+	var numDroppedEntries int
+	for numDroppedEntries < len(allLogData)-1 && len(allLogDataBytes) > maxTotalBytes {
+		numDroppedEntries++
+
+		allLogDataBytes, err = json.Marshal(allLogData[numDroppedEntries:])
+		if err != nil {
+			return nil, numDroppedEntries, err
+		}
+	}
+
+	return allLogDataBytes, numDroppedEntries, nil
+}

riverlog/river_log_test.go

@@ -270,6 +270,96 @@ func TestMiddleware(t *testing.T) {
 		)
 	})
 
+	t.Run("PrunesOldestEntriesAtMaxTotalBytes", func(t *testing.T) {
+		t.Parallel()
+
+		maxTotalBytes, err := json.Marshal([]logAttempt{
+			{Attempt: 1, Log: `msg="Logged from worker"` + "\n"},
+			{Attempt: 2, Log: `msg="Logged from worker"` + "\n"},
+		})
+		require.NoError(t, err)
+
+		testWorker, bundle := setup(t, &MiddlewareConfig{
+			MaxTotalBytes: len(maxTotalBytes),
+		})
+
+		workRes, err := testWorker.Work(ctx, t, bundle.tx, loggingArgs{Message: "Logged from worker"}, nil)
+		require.NoError(t, err)
+
+		// Set state back to available and unfinalize the job to make it runnable again.
+		workRes.Job, err = bundle.driver.UnwrapExecutor(bundle.tx).JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{
+			ID:                  workRes.Job.ID,
+			FinalizedAtDoUpdate: true,
+			FinalizedAt:         nil,
+			StateDoUpdate:       true,
+			State:               rivertype.JobStateAvailable,
+		})
+		require.NoError(t, err)
+
+		workRes, err = testWorker.WorkJob(ctx, t, bundle.tx, workRes.Job)
+		require.NoError(t, err)
+
+		// Set state back to available and unfinalize the job to make it runnable again.
+		workRes.Job, err = bundle.driver.UnwrapExecutor(bundle.tx).JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{
+			ID:                  workRes.Job.ID,
+			FinalizedAtDoUpdate: true,
+			FinalizedAt:         nil,
+			StateDoUpdate:       true,
+			State:               rivertype.JobStateAvailable,
+		})
+		require.NoError(t, err)
+
+		workRes, err = testWorker.WorkJob(ctx, t, bundle.tx, workRes.Job)
+		require.NoError(t, err)
+
+		var metadataWithLog metadataWithLog
+		require.NoError(t, json.Unmarshal(workRes.Job.Metadata, &metadataWithLog))
+
+		require.Equal(t, []logAttempt{
+			{Attempt: 2, Log: `msg="Logged from worker"` + "\n"},
+			{Attempt: 3, Log: `msg="Logged from worker"` + "\n"},
+		}, metadataWithLog.RiverLog)
+	})
+
+	t.Run("RetainsLatestEntryWhenTotalLimitTiny", func(t *testing.T) {
+		t.Parallel()
+
+		testWorker, bundle := setup(t, &MiddlewareConfig{
+			MaxTotalBytes: 1,
+		})
+
+		workRes, err := testWorker.Work(ctx, t, bundle.tx, loggingArgs{Message: "Logged from worker"}, nil)
+		require.NoError(t, err)
+
+		// Set state back to available and unfinalize the job to make it runnable again.
+		workRes.Job, err = bundle.driver.UnwrapExecutor(bundle.tx).JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{
+			ID:                  workRes.Job.ID,
+			FinalizedAtDoUpdate: true,
+			FinalizedAt:         nil,
+			StateDoUpdate:       true,
+			State:               rivertype.JobStateAvailable,
+		})
+		require.NoError(t, err)
+
+		workRes, err = testWorker.WorkJob(ctx, t, bundle.tx, workRes.Job)
+		require.NoError(t, err)
+
+		var metadataWithLog metadataWithLog
+		require.NoError(t, json.Unmarshal(workRes.Job.Metadata, &metadataWithLog))
+
+		require.Equal(t, []logAttempt{
+			{Attempt: 2, Log: `msg="Logged from worker"` + "\n"},
+		}, metadataWithLog.RiverLog)
+	})
+
+	t.Run("DefaultsMaxTotalBytes", func(t *testing.T) {
+		t.Parallel()
+
+		_, bundle := setup(t, nil)
+
+		require.Equal(t, maxTotalBytes, bundle.middleware.config.MaxTotalBytes)
+	})
+
 	t.Run("RawMiddleware", func(t *testing.T) {
 		t.Parallel()