diff --git a/galog.go b/galog.go index 7726637..6d456c3 100644 --- a/galog.go +++ b/galog.go @@ -593,15 +593,45 @@ func (lg *logger) Shutdown(timeout time.Duration) { ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() - // Flush any pending writes. + var wg sync.WaitGroup for _, queue := range backends { - if ctx.Err() != nil { - break - } - queue.entriesMutex.Lock() - lg.flushEnqueuedEntriesLocked(ctx, queue) - queue.entriesMutex.Unlock() - queue.backend.Shutdown(ctx) + wg.Add(1) + go func(bq *BackendQueue) { + defer wg.Done() + + // Attempt to acquire the lock to flush remaining entries. + // We poll TryLock with a timeout to prevent deadlocking if the + // background flusher is stuck on a blocked network write. + const pollInterval = 10 * time.Millisecond + var locked bool + for { + if bq.entriesMutex.TryLock() { + locked = true + break + } + if ctx.Err() != nil { + break + } + time.Sleep(pollInterval) + } + + if locked { + lg.flushEnqueuedEntriesLocked(ctx, bq) + bq.entriesMutex.Unlock() + } + bq.backend.Shutdown(ctx) + }(queue) + } + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-ctx.Done(): } } diff --git a/galog_deadlock_test.go b/galog_deadlock_test.go new file mode 100644 index 0000000..5404145 --- /dev/null +++ b/galog_deadlock_test.go @@ -0,0 +1,98 @@ +package galog + +import ( + "context" + "fmt" + "sync/atomic" + "testing" + "time" +) + +type blockingBackend struct { + id string + config Config + blockCh chan struct{} + logCount int32 + logCalled chan struct{} +} + +func newBlockingBackend(id string) *blockingBackend { + return &blockingBackend{ + id: id, + config: newBackendConfig(10), + blockCh: make(chan struct{}), + logCalled: make(chan struct{}), + } +} + +func (b *blockingBackend) ID() string { + return b.id +} + +func (b *blockingBackend) Log(entry *LogEntry) error { + count := atomic.AddInt32(&b.logCount, 1) + println(fmt.Sprintf("blockingBackend.Log called! count=%d", count)) + + if count == 1 { + // First call: return error to force enqueuing + println("First log: returning error to force queueing.") + return fmt.Errorf("simulated write failure") + } + + // Second call (from flush): block forever + println("Second log (flush): blocking forever.") + close(b.logCalled) + <-b.blockCh + println("blockingBackend.Log unblocked!") + return nil +} + +func (b *blockingBackend) Config() Config { + return b.config +} + +func (b *blockingBackend) Shutdown(ctx context.Context) error { + println("blockingBackend.Shutdown called!") + return nil +} + +func TestShutdownDeadlock(t *testing.T) { + tl := newTestLogger() + tl.SetLevel(DebugLevel) + + globalLogger := defaultLogger + defaultLogger = tl + t.Cleanup(func() { + defaultLogger = globalLogger + }) + + backend := newBlockingBackend("blocking-backend") + RegisterBackend(context.Background(), backend) + + // Log a message to trigger background writer goroutine + Debugf("This log will fail first and then block on flush") + + // Wait for Log to be called and block on the second attempt + select { + case <-backend.logCalled: + t.Log("Confirmed: Log was called twice and is now blocking on flush.") + case <-time.After(2 * time.Second): + t.Fatal("Fatal: Log did not block on flush within 2s!") + } + + shutdownDone := make(chan struct{}) + go func() { + println("Calling Shutdown...") + // Call Shutdown with a short timeout + Shutdown(100 * time.Millisecond) + println("Shutdown returned!") + close(shutdownDone) + }() + + select { + case <-shutdownDone: + t.Log("Shutdown completed successfully within timeout!") + case <-time.After(2 * time.Second): + t.Fatal("DEADLOCK DETECTED: Shutdown did not complete within 2s (timeout parameter was 100ms)") + } +}