rohit/probo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix worker retry

Browse Source 
Signed-off-by: Bryan Frimin <bryan@getprobo.com>
This commit is contained in:
Bryan Frimin
2026-02-20 13:55:20 +01:00
parent 9f54568625
commit ef28350c63
2 changed files with 10 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/coredata/electronic_signature.go
View File

@@ -268,6 +268,7 @@ FROM electronic_signatures
WHERE status = 'COMPLETED' WHERE status = 'COMPLETED'
AND certificate_file_id IS NULL AND certificate_file_id IS NULL
AND certificate_processing_started_at IS NULL AND certificate_processing_started_at IS NULL
AND attempt_count < max_attempts
ORDER BY signed_at ASC ORDER BY signed_at ASC
LIMIT 1 LIMIT 1
FOR UPDATE SKIP LOCKED FOR UPDATE SKIP LOCKED
@@ -362,6 +363,7 @@ WHERE status = 'COMPLETED'
AND certificate_file_id IS NULL AND certificate_file_id IS NULL
AND certificate_processing_started_at IS NOT NULL AND certificate_processing_started_at IS NOT NULL
AND certificate_processing_started_at < NOW() - $1::interval AND certificate_processing_started_at < NOW() - $1::interval
AND attempt_count < max_attempts
` `
_, err := conn.Exec(ctx, q, staleAfter) _, err := conn.Exec(ctx, q, staleAfter)
if err != nil { if err != nil {

8
pkg/esign/completion_certificate_worker.go
View File

@@ -151,6 +151,8 @@ func (w *CompletionCertificateWorker) processNext(ctx context.Context, sem chan
} }
scope := coredata.NewScopeFromObjectID(signature.ID) scope := coredata.NewScopeFromObjectID(signature.ID)
signature.CertificateProcessingStartedAt = &now signature.CertificateProcessingStartedAt = &now
signature.AttemptCount++
signature.LastAttemptedAt = &now
signature.UpdatedAt = now signature.UpdatedAt = now
if err := signature.Update(nonCancelableCtx, tx, scope); err != nil { if err := signature.Update(nonCancelableCtx, tx, scope); err != nil {
@@ -353,9 +355,15 @@ func (w *CompletionCertificateWorker) handleCertFailure(
return w.pg.WithTx( return w.pg.WithTx(
ctx, ctx,
func(tx pg.Conn) error { func(tx pg.Conn) error {
errStr := processingError.Error()
signature.LastError = &errStr
signature.CertificateProcessingStartedAt = nil signature.CertificateProcessingStartedAt = nil
signature.UpdatedAt = time.Now() signature.UpdatedAt = time.Now()
if signature.AttemptCount >= signature.MaxAttempts {
signature.Status = coredata.ElectronicSignatureStatusFailed
}
if err := signature.Update(ctx, tx, scope); err != nil { if err := signature.Update(ctx, tx, scope); err != nil {
return fmt.Errorf("cannot update signature: %w", err) return fmt.Errorf("cannot update signature: %w", err)
} }