riverlog: use sjson for log append and pruning

The raw-message iteration still rebuilt log arrays through json
marshaling, which left append performance significantly worse than the
baseline implementation used on master.

Switch riverlog history to gjson/sjson so operations stay on raw JSON
arrays. The middleware reads only `river:log` with `gjson`, appends with
`sjson.SetRawBytes` using `-1`, and enforces `MaxTotalBytes` by deleting
oldest entries with `sjson.DeleteBytes("0")`.

Comparison for `BenchmarkMiddlewareWorkAppend` (`-benchtime=100ms`):

    case: 256 KB
    impl           ns/op      B/op  allocs/op
    baseline      976052    645856        291
    rawmessage   1605106    698233        293
    sjson         614433   1167265         25

    case: 2 MB
    impl           ns/op      B/op  allocs/op
    baseline     7762435   5920954       2096
    rawmessage  12516198   6651551       2098
    sjson        4485274   6482082         32

The pruning test changed only to avoid a brittle exact-byte boundary.
`MaxTotalBytes` is now set slightly above the two-entry payload so the
test consistently verifies the intended behavior: appending a third
entry drops the oldest and keeps the newest two.

Author: bgentry

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### 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).
+- Improved `riverlog` performance and reduced memory amplification when appending to large persisted `river:log` histories. [PR #1157](https://github.com/riverqueue/river/pull/1157).
 
 ## [0.31.0] - 2026-02-21
 

riverlog/river_log.go

@@ -8,9 +8,13 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 
+	"github.com/tidwall/gjson"
+	"github.com/tidwall/sjson"
+
 	"github.com/riverqueue/river/internal/jobexecutor"
 	"github.com/riverqueue/river/rivershared/baseservice"
 	"github.com/riverqueue/river/rivertype"
@@ -162,15 +166,8 @@ type metadataWithLog struct {
 	RiverLog []logAttempt `json:"river:log"`
 }
 
-type metadataWithRawLog struct {
-	RiverLog []json.RawMessage `json:"river:log"`
-}
-
 func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner func(context.Context) error) error {
-	var (
-		existingRawLogData metadataWithRawLog
-		logBuf             bytes.Buffer
-	)
+	var logBuf bytes.Buffer
 
 	switch {
 	case m.newCustomContext != nil:
@@ -182,10 +179,6 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu
 		return errors.New("expected either newContextLogger or newSlogHandler to be set")
 	}
 
-	if err := json.Unmarshal(job.Metadata, &existingRawLogData); err != nil {
-		return err
-	}
-
 	metadataUpdates, hasMetadataUpdates := jobexecutor.MetadataUpdatesFromWorkContext(ctx)
 	if !hasMetadataUpdates {
 		return errors.New("expected to find metadata updates in context, but didn't")
@@ -221,7 +214,7 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu
 			return
 		}
 
-		allLogDataBytes, numDroppedEntries, err := marshalRawLogDataWithCap(append(existingRawLogData.RiverLog, json.RawMessage(newLogEntryBytes)), m.config.MaxTotalBytes)
+		allLogDataBytes, numDroppedEntries, err := appendLogDataWithCap(job.Metadata, newLogEntryBytes, m.config.MaxTotalBytes)
 		if err != nil {
 			m.Logger.ErrorContext(ctx, m.Name+": Error marshaling log data",
 				slog.Any("error", err),
@@ -242,8 +235,24 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu
 	return doInner(ctx)
 }
 
-func marshalRawLogDataWithCap(allLogData []json.RawMessage, maxTotalBytes int) ([]byte, int, error) {
-	allLogDataBytes, err := json.Marshal(allLogData)
+func appendLogDataWithCap(metadataBytes, newLogEntryBytes []byte, maxTotalBytes int) ([]byte, int, error) {
+	if !json.Valid(metadataBytes) {
+		return nil, 0, errors.New("metadata is not valid JSON")
+	}
+
+	existingLogData := gjson.GetBytes(metadataBytes, metadataKey)
+	var allLogDataBytes []byte
+	switch {
+	case !existingLogData.Exists():
+		allLogDataBytes = []byte("[]")
+	case existingLogData.IsArray():
+		allLogDataBytes = []byte(existingLogData.Raw)
+	default:
+		return nil, 0, fmt.Errorf("%q value is not an array", metadataKey)
+	}
+
+	var err error
+	allLogDataBytes, err = sjson.SetRawBytes(allLogDataBytes, "-1", newLogEntryBytes)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -253,14 +262,15 @@ func marshalRawLogDataWithCap(allLogData []json.RawMessage, maxTotalBytes int) (
 	}
 
 	// Drop oldest entries first, while always retaining the latest one.
+	numEntries := len(gjson.ParseBytes(allLogDataBytes).Array())
 	var numDroppedEntries int
-	for numDroppedEntries < len(allLogData)-1 && len(allLogDataBytes) > maxTotalBytes {
-		numDroppedEntries++
-
-		allLogDataBytes, err = json.Marshal(allLogData[numDroppedEntries:])
+	for numEntries > 1 && len(allLogDataBytes) > maxTotalBytes {
+		allLogDataBytes, err = sjson.DeleteBytes(allLogDataBytes, "0")
 		if err != nil {
 			return nil, numDroppedEntries, err
 		}
+		numEntries--
+		numDroppedEntries++
 	}
 
 	return allLogDataBytes, numDroppedEntries, nil

riverlog/river_log_test.go

@@ -280,7 +280,8 @@ func TestMiddleware(t *testing.T) {
 		require.NoError(t, err)
 
 		testWorker, bundle := setup(t, &MiddlewareConfig{
-			MaxTotalBytes: len(maxTotalBytes),
+			// Keep two entries, force pruning once a third is appended.
+			MaxTotalBytes: len(maxTotalBytes) + 16,
 		})
 
 		workRes, err := testWorker.Work(ctx, t, bundle.tx, loggingArgs{Message: "Logged from worker"}, nil)