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

add dbgs

Browse source
This commit is contained in:
Pranav Shridhar 2024-04-09 12:22:09 +05:30
parent fd7e5cc07b
commit 330893cc06
1 changed files with 16 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
backends/sqlite/sqlite_backend.go
View file

@ -323,9 +323,18 @@ func (s *SqliteBackend) handleJob(ctx context.Context, jobID string) (err error)
return handler.ErrNoHandlerForQueue
}
fmt.Printf("Current job : %+v", job)
ctx = withJobContext(ctx, job)
fmt.Printf("Updated ctx with job : %+v", ctx)
var tx *sql.Tx
s.dbMutex.Lock()
tx, err = s.db.Begin()
if err != nil {
return
}
ctx = context.WithValue(ctx, txCtxVarKey, tx)
fmt.Printf("Updated ctx with tx : %+v", ctx)
err = s.updateJob(ctx, jobErr, internal.JobStatusInProgress)
if err != nil {
if errors.Is(err, context.Canceled) {
@ -334,13 +343,18 @@ func (s *SqliteBackend) handleJob(ctx context.Context, jobID string) (err error)
err = fmt.Errorf("error updating job status: %w", err)
return err
}
err = tx.Commit()
if err != nil {
errMsg := "unable to commit job transaction. retrying this job may duplicate work:"
s.logger.Error(errMsg, slog.String("queue", h.Queue), slog.Any("error", err), slog.Int64("job_id", job.ID))
_ = tx.Rollback()
return fmt.Errorf("%s %w", errMsg, err)
}
s.dbMutex.Unlock()
// execute the queue handler of this job
jobErr = handler.Exec(ctx, h)
var tx *sql.Tx
s.dbMutex.Lock()
defer s.dbMutex.Unlock()