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

Fix ctx cancel not listen

Browse Source 
Signed-off-by: Bryan Frimin <bryan@getprobo.com>
This commit is contained in:
Bryan Frimin
2026-02-20 12:55:14 +01:00
parent c6e198854e
commit 28d0d435d2
2 changed files with 23 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
pkg/esign/completion_certificate_worker.go
View File

@@ -115,12 +115,11 @@ LOOP:
case <-time.After(w.interval): case <-time.After(w.interval):
// From there we should not accept cancelations anymore. // From there we should not accept cancelations anymore.
nonCancelableCtx := context.WithoutCancel(ctx) nonCancelableCtx := context.WithoutCancel(ctx)
w.recoverStaleCertificateRows(nonCancelableCtx) w.recoverStaleCertificateRows(nonCancelableCtx)
for { for {
if err := w.processNext(nonCancelableCtx, sem, &wg); err != nil { if err := w.processNext(ctx, sem, &wg); err != nil {
if !errors.Is(err, coredata.ErrResourceNotFound) { if !errors.Is(err, coredata.ErrResourceNotFound) {
w.logger.ErrorCtx(nonCancelableCtx, "cannot process certificate", log.Error(err)) w.logger.ErrorCtx(ctx, "cannot process certificate", log.Error(err))
} }
break break
} }
@@ -132,26 +131,29 @@ LOOP:
func (w *CompletionCertificateWorker) processNext(ctx context.Context, sem chan struct{}, wg *sync.WaitGroup) error { func (w *CompletionCertificateWorker) processNext(ctx context.Context, sem chan struct{}, wg *sync.WaitGroup) error {
select { select {
case sem <- struct{}{}: case sem <- struct{}{}:
case <-ctx.Done(): case <-ctx.Done(): // FIXME: this will never be fired
return ctx.Err() return ctx.Err()
} }
var ( var (
signature coredata.ElectronicSignature signature coredata.ElectronicSignature
now = time.Now() now = time.Now()
// From there we should not accept cancelations anymore.
nonCancelableCtx = context.WithoutCancel(ctx)
) )
if err := w.pg.WithTx( if err := w.pg.WithTx(
ctx, nonCancelableCtx,
func(tx pg.Conn) error { func(tx pg.Conn) error {
if err := signature.LoadNextCompletedWithoutCertificateForUpdate(ctx, tx); err != nil { if err := signature.LoadNextCompletedWithoutCertificateForUpdate(nonCancelableCtx, tx); err != nil {
return err return err
} }
scope := coredata.NewScopeFromObjectID(signature.ID) scope := coredata.NewScopeFromObjectID(signature.ID)
signature.CertificateProcessingStartedAt = &now signature.CertificateProcessingStartedAt = &now
signature.UpdatedAt = now signature.UpdatedAt = now
if err := signature.Update(ctx, tx, scope); err != nil { if err := signature.Update(nonCancelableCtx, tx, scope); err != nil {
return fmt.Errorf("cannot update signature: %w", err) return fmt.Errorf("cannot update signature: %w", err)
} }
@@ -169,9 +171,9 @@ func (w *CompletionCertificateWorker) processNext(ctx context.Context, sem chan
scope := coredata.NewScopeFromObjectID(signature.ID) scope := coredata.NewScopeFromObjectID(signature.ID)
if err := w.generateAndCommit(ctx, &signature); err != nil { if err := w.generateAndCommit(nonCancelableCtx, &signature); err != nil {
if err := w.handleCertFailure(ctx, &signature, scope, err); err != nil { if err := w.handleCertFailure(nonCancelableCtx, &signature, scope, err); err != nil {
w.logger.ErrorCtx(ctx, "cannot handle certificate failure", log.Error(err)) w.logger.ErrorCtx(nonCancelableCtx, "cannot handle certificate failure", log.Error(err))
} }
} }
}(signature) }(signature)

19
pkg/esign/sealing_worker.go
View File

@@ -114,10 +114,10 @@ LOOP:
case <-time.After(w.interval): case <-time.After(w.interval):
// From there we should not accept cancelations anymore. // From there we should not accept cancelations anymore.
nonCancelableCtx := context.WithoutCancel(ctx) nonCancelableCtx := context.WithoutCancel(ctx)
w.recoverStaleRows(nonCancelableCtx) w.recoverStaleRows(nonCancelableCtx)
for { for {
if err := w.processNext(nonCancelableCtx, sem, &wg); err != nil { if err := w.processNext(ctx, sem, &wg); err != nil {
if !errors.Is(err, coredata.ErrResourceNotFound) { if !errors.Is(err, coredata.ErrResourceNotFound) {
w.logger.ErrorCtx(nonCancelableCtx, "cannot claim signature", log.Error(err)) w.logger.ErrorCtx(nonCancelableCtx, "cannot claim signature", log.Error(err))
} }
@@ -139,12 +139,15 @@ func (w *SealingWorker) processNext(ctx context.Context, sem chan struct{}, wg *
var ( var (
signature = coredata.ElectronicSignature{} signature = coredata.ElectronicSignature{}
now = time.Now() now = time.Now()
// From there we should not accept cancelations anymore.
nonCancelableCtx = context.WithoutCancel(ctx)
) )
if err := w.pg.WithTx( if err := w.pg.WithTx(
ctx, nonCancelableCtx,
func(tx pg.Conn) error { func(tx pg.Conn) error {
if err := signature.LoadNextAcceptedForUpdateSkipLocked(ctx, tx); err != nil { if err := signature.LoadNextAcceptedForUpdateSkipLocked(nonCancelableCtx, tx); err != nil {
return err return err
} }
@@ -153,7 +156,7 @@ func (w *SealingWorker) processNext(ctx context.Context, sem chan struct{}, wg *
signature.AttemptCount++ signature.AttemptCount++
signature.LastAttemptedAt = &now signature.LastAttemptedAt = &now
signature.UpdatedAt = now signature.UpdatedAt = now
if err := signature.Update(ctx, tx, coredata.NewNoScope()); err != nil { if err := signature.Update(nonCancelableCtx, tx, coredata.NewNoScope()); err != nil {
return fmt.Errorf("cannot update signature: %w", err) return fmt.Errorf("cannot update signature: %w", err)
} }
@@ -169,9 +172,9 @@ func (w *SealingWorker) processNext(ctx context.Context, sem chan struct{}, wg *
defer wg.Done() defer wg.Done()
defer func() { <-sem }() defer func() { <-sem }()
if err := w.sealAndCommit(ctx, &signature); err != nil { if err := w.sealAndCommit(nonCancelableCtx, &signature); err != nil {
if err := w.failSignature(ctx, &signature, err); err != nil { if err := w.failSignature(nonCancelableCtx, &signature, err); err != nil {
w.logger.ErrorCtx(ctx, "cannot fail signature", log.Error(err)) w.logger.ErrorCtx(nonCancelableCtx, "cannot fail signature", log.Error(err))
} }
} }
}(signature) }(signature)