fix: finish queue cleanup before stop returns
This commit is contained in:
@@ -226,6 +226,7 @@ func (q *Queue) Stop(contexts ...context.Context) error {
|
|||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
if shutdownOwner && workCancel != nil {
|
if shutdownOwner && workCancel != nil {
|
||||||
workCancel()
|
workCancel()
|
||||||
|
<-done
|
||||||
}
|
}
|
||||||
return ctx.Err()
|
return ctx.Err()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -393,7 +393,7 @@ func TestQueueByteBudgetTracksOwnedEntrySnapshot(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestQueueCanceledStopEventuallyReleasesAllBudget(t *testing.T) {
|
func TestQueueCanceledStopReleasesAllBudgetBeforeReturning(t *testing.T) {
|
||||||
entry := &logger.LogEntry{RequestID: "queued"}
|
entry := &logger.LogEntry{RequestID: "queued"}
|
||||||
q := logger.NewQueue(nil, 32, 32, 1, time.Hour, 32*logger.EstimatedBytes(entry))
|
q := logger.NewQueue(nil, 32, 32, 1, time.Hour, 32*logger.EstimatedBytes(entry))
|
||||||
q.Start(context.Background())
|
q.Start(context.Background())
|
||||||
@@ -405,10 +405,6 @@ func TestQueueCanceledStopEventuallyReleasesAllBudget(t *testing.T) {
|
|||||||
cancel()
|
cancel()
|
||||||
_ = q.Stop(ctx)
|
_ = q.Stop(ctx)
|
||||||
|
|
||||||
deadline := time.Now().Add(time.Second)
|
|
||||||
for q.Stats().Bytes != 0 && time.Now().Before(deadline) {
|
|
||||||
time.Sleep(time.Millisecond)
|
|
||||||
}
|
|
||||||
if stats := q.Stats(); stats.Bytes != 0 || stats.Failed != 32 {
|
if stats := q.Stats(); stats.Bytes != 0 || stats.Failed != 32 {
|
||||||
t.Fatalf("stats=%+v", stats)
|
t.Fatalf("stats=%+v", stats)
|
||||||
}
|
}
|
||||||
@@ -540,10 +536,6 @@ func TestQueueStopDeadlineCancelsBlockedSend(t *testing.T) {
|
|||||||
if err := q.Stop(ctx); !errors.Is(err, context.DeadlineExceeded) {
|
if err := q.Stop(ctx); !errors.Is(err, context.DeadlineExceeded) {
|
||||||
t.Fatalf("Stop error=%v want deadline exceeded", err)
|
t.Fatalf("Stop error=%v want deadline exceeded", err)
|
||||||
}
|
}
|
||||||
deadline := time.Now().Add(250 * time.Millisecond)
|
|
||||||
for q.Stats().Bytes != 0 && time.Now().Before(deadline) {
|
|
||||||
time.Sleep(time.Millisecond)
|
|
||||||
}
|
|
||||||
if stats := q.Stats(); stats.Bytes != 0 || stats.Ambiguous != 1 || stats.Failed != 1 {
|
if stats := q.Stats(); stats.Bytes != 0 || stats.Ambiguous != 1 || stats.Failed != 1 {
|
||||||
t.Fatalf("stats after canceled Send=%+v", stats)
|
t.Fatalf("stats after canceled Send=%+v", stats)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user