Periodic Jobs question

[sheldondz]

We are trying the new dynamic Periodic jobs feature in 0.6.0, and are facing the following issues

Schedule a job to run every day at 8AM, base on timezone

schedule, err := cron.ParseStandard("CRON_TZ=Asia/Calcutta  0 8 * * *")
	if err != nil {
		r.logger.Error("failed to add calendar event", log.Error(err))
	}
	handle := r.jobManager.Client.PeriodicJobs().Add(
		river.NewPeriodicJob(
			schedule,
			func() (river.JobArgs, *river.InsertOpts) {
				return domain.CalendarEventJobArgs{
					EventType:      "test event"
					EventId:       1
				}, nil
			},
			nil,
		),
	)

After this code is executed there is no entry added in the river_job table, if we set RunOnStart we see an entry and the worker also executes but after that the periodic jobs does not repeat.

We also tried a simple test with

periodicJobs := []*river.PeriodicJob{
		river.NewPeriodicJob(
			river.PeriodicInterval(15*time.Second),
			func() (river.JobArgs, *river.InsertOpts) {
				return domain.CalendarEventJobArgs{}, nil
			},
			&river.PeriodicJobOpts{RunOnStart: true},
		),
	}

	riverClient, err := river.NewClient(riverpgxv5.New(s.storage.Pool), &river.Config{
		Queues: map[string]river.QueueConfig{
			river.QueueDefault: {MaxWorkers: 100},
		},
		Workers:      workers,
		PeriodicJobs: periodicJobs,
	})
	if err != nil {
		s.logger.Error("failed to create river client", log.Error(err))
		return nil, err
	}

	// Run the client inline. All executed jobs will inherit from jobCtx:
	if err := riverClient.Start(ctx); err != nil {
		// handle error
		s.logger.Error("failed to start river client", log.Error(err))
		return nil, err
	}

Here too an entry is added in the DB the worker runs immediately, after that the worker does not run again after 15 sec. Is there something i am missing here?

[brandur]

Here's a fully functional code sample based on your code above:

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/riverqueue/river"
	"github.com/riverqueue/river/riverdriver/riverpgxv5"
)

type CalendarEventJobArgs struct{}

func (CalendarEventJobArgs) Kind() string { return "calendar_event" }

type CalendarEventWorker struct {
	// An embedded WorkerDefaults sets up default methods to fulfill the rest of
	// the Worker interface:
	river.WorkerDefaults[CalendarEventJobArgs]
}

func (w *CalendarEventWorker) Work(ctx context.Context, job *river.Job[CalendarEventJobArgs]) error {
	fmt.Printf("%s: CalendarEvented ran\n", time.Now().Format(time.DateTime))
	return nil
}

func doStart(ctx context.Context, logger *slog.Logger) (*struct{}, error) {
	dbPool, err := pgxpool.New(ctx, "postgres://localhost/river-periodic-job-test")
	if err != nil {
		return nil, err
	}

	periodicJobs := []*river.PeriodicJob{
		river.NewPeriodicJob(
			river.PeriodicInterval(15*time.Second),
			func() (river.JobArgs, *river.InsertOpts) {
				return CalendarEventJobArgs{}, nil
			},
			&river.PeriodicJobOpts{RunOnStart: true},
		),
	}

	workers := river.NewWorkers()
	// AddWorker panics if the worker is already registered or invalid:
	river.AddWorker(workers, &CalendarEventWorker{})

	riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
		Queues: map[string]river.QueueConfig{
			river.QueueDefault: {MaxWorkers: 100},
		},
		Workers:      workers,
		PeriodicJobs: periodicJobs,
	})
	if err != nil {
		logger.Error("failed to create river client", "err", err)
		return nil, err
	}

	// Run the client inline. All executed jobs will inherit from jobCtx:
	if err := riverClient.Start(ctx); err != nil {
		// handle error
		logger.Error("failed to start river client", "err", err)
		return nil, err
	}

	return nil, nil
}

func main() {
	ctx := context.Background()
	logger := slog.New(slog.NewTextHandler(os.Stdout, nil))

	if _, err := doStart(ctx, logger); err != nil {
		panic(err)
	}

	foreverChan := make(chan struct{})
	<-foreverChan
}

Run it, and you can see the job running every 15 seconds:

$ go run main.go
2024-05-17 11:03:22: CalendarEvented ran
2024-05-17 11:03:37: CalendarEvented ran
2024-05-17 11:03:52: CalendarEvented ran
2024-05-17 11:04:07: CalendarEvented ran
2024-05-17 11:04:22: CalendarEvented ran
2024-05-17 11:04:37: CalendarEvented ran
2024-05-17 11:04:52: CalendarEvented ran
2024-05-17 11:05:07: CalendarEvented ran
2024-05-17 11:05:22: CalendarEvented ran
2024-05-17 11:05:37: CalendarEvented ran
2024-05-17 11:05:52: CalendarEvented ran
2024-05-17 11:06:07: CalendarEvented ran
2024-05-17 11:06:22: CalendarEvented ran
2024-05-17 11:06:37: CalendarEvented ran

Here's the same thing in the database:

river-periodic-job-test=# select kind, state, finalized_at from river_job order by id;
      kind      |   state   |         finalized_at
----------------+-----------+-------------------------------
 calendar_event | completed | 2024-05-17 02:02:05.964429-07
 calendar_event | completed | 2024-05-17 02:02:20.959239-07
 calendar_event | completed | 2024-05-17 02:02:35.958579-07
 calendar_event | completed | 2024-05-17 02:02:50.956588-07
 calendar_event | completed | 2024-05-17 02:03:05.958406-07
 calendar_event | completed | 2024-05-17 02:03:22.724301-07
 calendar_event | completed | 2024-05-17 02:03:37.725189-07
 calendar_event | completed | 2024-05-17 02:03:52.723318-07
 calendar_event | completed | 2024-05-17 02:04:07.724498-07
 calendar_event | completed | 2024-05-17 02:04:22.72267-07
 calendar_event | completed | 2024-05-17 02:04:37.724326-07

Based on code not shown in your sample above:

1. Is your job unique? If so, it may be excluded from periodic insert because a conforming job is already in the database.
2. Does your program keep running after Client.Start's been invoked? If it ends too soon, obviously jobs won't be inserted.

[sheldondz]

@brandur Thanks for the detailed code sample, i think its because my job is unique its failing to run again. Will check this and update.

[sheldondz]

I can confirm that removing the uniqueOpts it works as expected. Thanks.

[brandur]

You got it!