Periodic job next_run_at can advance after job insert failure

Summary

PeriodicJobEnqueuer.insertBatch starts a transaction, calls the configured
periodic job insert function, and then upserts durable periodic job
next_run_at records.

If the insert function returns an error, insertBatch logs the error but still
continues to PeriodicJobUpsertMany and can commit the transaction. For insert
errors that happen before any database write, the transaction is still usable,
so the durable next-run record can advance even though no job was inserted for
the due occurrence.

Confirmed against River master at
4aa884e927fc07635605967729a288ac9416ebf1 on 2026-07-10 with Go 1.26.5
on Linux. I found related periodic-job discussions and docs, but not an exact
public report for this insert-error/next-run advancement case.

Why this seems wrong

• The durable next_run_at record says the due occurrence has been accounted
  for, but the corresponding job row may not exist.
• A later leader or restarted process can observe the advanced durable record
  and skip the missed occurrence.
• Insert callbacks can fail before touching the database, for example due to
  validation, middleware, or application insertion logic.
• The transaction already groups the insert and the next-run upsert, so
  aborting on insert error preserves the natural all-or-nothing boundary.

Relevant code

In internal/maintenance/periodic_job_enqueuer.go, insertBatch logs the
insert error but then falls through to the durable next-run upsert:

if len(insertParamsMany) > 0 {
    if _, err := s.Config.Insert(ctx, tx, insertParamsMany); err != nil {
        s.Logger.ErrorContext(ctx, s.Name+": Error inserting periodic jobs",
            "error", err.Error(), "num_jobs", len(insertParamsMany))
    }
}

if len(periodicJobUpsertParams.Jobs) > 0 {
    if _, err = s.Config.Pilot.PeriodicJobUpsertMany(ctx, tx, periodicJobUpsertParams); err != nil {
        s.Logger.ErrorContext(ctx, s.Name+": Error upserting periodic job next run times",
            "error", err.Error(), "num_jobs", len(insertParamsMany), "num_next_run_at_upserts", len(periodicJobUpsertParams.Jobs))
        return
    }
}

Reproduction

Add a focused test to internal/maintenance/periodic_job_enqueuer_test.go
using an insert callback that fails before touching the transaction and a pilot
mock that records whether the durable upsert was called:

t.Run("DoesNotAdvanceDurableNextRunAfterInsertError", func(t *testing.T) {
    t.Parallel()

    exec := &periodicInsertBatchExecutorMock{}
    pilotMock := NewPilotPeriodicJobMock()

    svc, err := NewPeriodicJobEnqueuer(riversharedtest.BaseServiceArchetype(t), &PeriodicJobEnqueuerConfig{
        Insert: func(ctx context.Context, tx riverdriver.ExecutorTx, insertParams []*rivertype.JobInsertParams) ([]*rivertype.JobInsertResult, error) {
            return nil, errors.New("insert failed before database write")
        },
        Pilot:  pilotMock,
        Schema: "river_test_schema",
    }, exec)
    require.NoError(t, err)
    svc.TestSignals.Init(t)

    var upsertCalled bool
    pilotMock.PeriodicJobUpsertManyMock = func(ctx context.Context, exec riverdriver.Executor, params *riverpilot.PeriodicJobUpsertManyParams) ([]*riverpilot.PeriodicJob, error) {
        upsertCalled = true
        return nil, nil
    }

    now := time.Now().UTC()
    svc.insertBatch(ctx, []*rivertype.JobInsertParams{{Kind: "periodic_insert_error", Queue: rivercommon.QueueDefault, State: rivertype.JobStateAvailable}}, &riverpilot.PeriodicJobUpsertManyParams{
        Jobs: []*riverpilot.PeriodicJobUpsertParams{{ID: "periodic_insert_error", NextRunAt: now.Add(time.Minute), UpdatedAt: now}},
        Schema: "river_test_schema",
    })

    require.False(t, upsertCalled)
    require.False(t, exec.tx.committed)
    require.True(t, exec.tx.rolledBack)
})

Run:

go test ./internal/maintenance -run "TestPeriodicJobEnqueuer/DoesNotAdvanceDurableNextRunAfterInsertError" -count=1 -v

Current result:

level=ERROR msg="maintenance.PeriodicJobEnqueuer: Error inserting periodic jobs" error="insert failed before database write" num_jobs=1
Error:       Should be false
Test:        TestPeriodicJobEnqueuer/DoesNotAdvanceDurableNextRunAfterInsertError

The failing assertion checks that PeriodicJobUpsertMany was not called after
the insert callback failed.

Expected behavior

If inserting the due periodic job fails, the batch should not commit the next
next_run_at for that same occurrence.

Suggested fix

Return immediately after an insert error so the deferred rollback runs and the
next-run upsert is not committed:

if _, err := s.Config.Insert(ctx, tx, insertParamsMany); err != nil {
    s.Logger.ErrorContext(ctx, s.Name+": Error inserting periodic jobs",
        "error", err.Error(), "num_jobs", len(insertParamsMany))
    return
}

One follow-up consideration: the scheduler currently advances the in-memory
nextRunAt before insertBatch runs. If immediate retry in the same process
is desired after an insert failure, that in-memory advancement should also be
deferred until the batch commits or rolled back when the batch aborts.