me/neoq
1
0
Fork 0
mirror of https://github.com/acaloiaro/neoq synced 2026-07-21 10:12:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

add in progress job status

Browse source
This commit is contained in:
Pranav Shridhar 2024-04-09 09:48:16 +05:30
parent 4dea09d446
commit a575976aa7
2 changed files with 32 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
backends/sqlite/sqlite_backend.go
View file

@ -325,6 +325,17 @@ func (s *SqliteBackend) handleJob(ctx context.Context, jobID string) (err error)
ctx = withJobContext(ctx, job)
s.dbMutex.Lock()
err = s.updateJob(ctx, jobErr, internal.JobStatusInProgress)
if err != nil {
if errors.Is(err, context.Canceled) {
return
}
err = fmt.Errorf("error updating job status: %w", err)
return err
}
s.dbMutex.Unlock()
// execute the queue handler of this job
jobErr = handler.Exec(ctx, h)
@ -341,12 +352,11 @@ func (s *SqliteBackend) handleJob(ctx context.Context, jobID string) (err error)
ctx = context.WithValue(ctx, txCtxVarKey, tx)
err = s.updateJob(ctx, jobErr)
err = s.updateJob(ctx, jobErr, "")
if err != nil {
if errors.Is(err, context.Canceled) {
return
}
err = fmt.Errorf("error updating job status: %w", err)
return err
}
@ -394,16 +404,9 @@ func withJobContext(ctx context.Context, j *jobs.Job) context.Context {
return context.WithValue(ctx, internal.JobCtxVarKey, j)
}
func (s *SqliteBackend) updateJob(ctx context.Context, jobErr error) (err error) {
status := internal.JobStatusProcessed
func (s *SqliteBackend) updateJob(ctx context.Context, jobErr error, status string) (err error) {
errMsg := ""
if jobErr != nil {
s.logger.Error("job failed", slog.Any("job_error", jobErr))
status = internal.JobStatusFailed
errMsg = jobErr.Error()
}
var job *jobs.Job
if job, err = jobs.FromContext(ctx); err != nil {
return fmt.Errorf("error getting job from context: %w", err)
@ -419,6 +422,21 @@ func (s *SqliteBackend) updateJob(ctx context.Context, jobErr error) (err error)
return
}
// In progress job
if len(status) != 0 {
qstr := "UPDATE neoq_jobs SET status = $1, error = $2 WHERE id = $3"
_, err = tx.ExecContext(ctx, qstr, status, errMsg, job.ID)
return
}
status = internal.JobStatusProcessed
if jobErr != nil {
s.logger.Error("job failed", slog.Any("job_error", jobErr))
status = internal.JobStatusFailed
errMsg = jobErr.Error()
}
var runAfter time.Time
if status == internal.JobStatusFailed {
runAfter = internal.CalculateBackoff(job.Retries)

7
internal/internal.go
View file

@ -10,9 +10,10 @@ import (
type contextKey struct{}
const (
JobStatusNew = "new"
JobStatusProcessed = "processed"
JobStatusFailed = "failed"
JobStatusNew = "new"
JobStatusInProgress = "in progress"
JobStatusProcessed = "processed"
JobStatusFailed = "failed"
)
var JobCtxVarKey contextKey