1 files changed, 103 insertions, 19 deletions

diff --git a/internal/io/dlog/loggers/file.go b/internal/io/dlog/loggers/file.go
index 8e567bc..b3b6a10 100644
--- a/internal/io/dlog/loggers/file.go
+++ b/internal/io/dlog/loggers/file.go
@@ -12,6 +12,26 @@ import (
 	"github.com/mimecast/dtail/internal/config"
 )
 
+const (
+	// fileWriterBufSize is the size of the bufio buffer wrapping the log file
+	// descriptor. A real buffer (instead of the old 1-byte writer that forced a
+	// write syscall per line) lets bulk payload — e.g. dcat/dgrep tee — batch
+	// into ~one syscall per bufferful, cutting the client receive-path syscall
+	// count and CPU by ~5x. bufio auto-flushes when full, so high-throughput
+	// output never stalls in the buffer.
+	fileWriterBufSize = 64 * 1024
+	// fileIdleFlushInterval bounds how long buffered data may sit unwritten when
+	// output goes idle (follow/interactive mode trickling a few lines). Without
+	// it, low-volume output would be stuck behind the buffer until it fills or
+	// the logger shuts down, so follow/tail would appear frozen on disk.
+	fileIdleFlushInterval = 100 * time.Millisecond
+	// fileFlushTimeout bounds how long a synchronous Flush() waits for the
+	// logger goroutine to acknowledge. It exists purely as a deadlock guard for
+	// the rare case where the goroutine is paused or already gone (e.g. Flush
+	// racing shutdown); under normal operation the ack is near-instant.
+	fileFlushTimeout = 2 * time.Second
+)
+
 type fileMessageBuf struct {
 	now     time.Time
 	message string
@@ -23,7 +43,13 @@ type file struct {
 	pauseCh      chan struct{}
 	resumeCh     chan struct{}
 	rotateCh     chan struct{}
-	flushCh      chan struct{}
+	// flushCh carries a per-call reply channel so Flush() can block until the
+	// logger goroutine has actually drained the buffer channel and flushed the
+	// bufio writer to disk. This makes Flush() synchronous, which the crash path
+	// (dlog.FatalPanic -> Flush -> panic) relies on: an async signal could let
+	// the process unwind before the goroutine drains, dropping up to one buffer
+	// (64KB) of Fatal diagnostics.
+	flushCh      chan chan struct{}
 	fd           *os.File
 	writer       *bufio.Writer
 	mutex        sync.Mutex
@@ -32,13 +58,20 @@ type file struct {
 	strategy     Strategy
 }
 
+var _ Logger = (*file)(nil)
+
 func newFile(strategy Strategy) *file {
+	// Pause/Resume/Rotate use capacity-1, non-blocking coalescing sends so
+	// callers never block on the logger goroutine (repeated signals collapse
+	// into one pending notification). flushCh is unbuffered and carries a reply
+	// channel because Flush() is synchronous: it must wait for the goroutine to
+	// drain and write before returning.
 	return &file{
 		bufferCh: make(chan *fileMessageBuf, runtime.NumCPU()*100),
-		pauseCh:  make(chan struct{}),
-		resumeCh: make(chan struct{}),
-		rotateCh: make(chan struct{}),
-		flushCh:  make(chan struct{}),
+		pauseCh:  make(chan struct{}, 1),
+		resumeCh: make(chan struct{}, 1),
+		rotateCh: make(chan struct{}, 1),
+		flushCh:  make(chan chan struct{}),
 		strategy: strategy,
 	}
 }
@@ -67,17 +100,43 @@ func (f *file) Start(ctx context.Context, wg *sync.WaitGroup) {
 
 	go func() {
 		defer wg.Done()
+		// Idle-flush ticker: with a real (64KB) buffer, low-volume output
+		// (follow/interactive) would otherwise sit in the buffer until it
+		// fills. The ticker flushes any partial buffer promptly so follow/tail
+		// output reaches disk within fileIdleFlushInterval. flush() is cheap
+		// when nothing is buffered.
+		ticker := time.NewTicker(fileIdleFlushInterval)
+		defer ticker.Stop()
 		for {
 			select {
 			case m := <-f.bufferCh:
 				f.write(m)
+			case <-ticker.C:
+				f.flush()
 			case <-f.pauseCh:
+				// Flush before pausing so all output produced so far is on
+				// disk before the caller (e.g. an interactive prompt) writes
+				// directly to the terminal/file; preserves ordering.
+				f.flush()
 				pause(ctx)
-			case <-f.flushCh:
+			case done := <-f.flushCh:
+				// Synchronous flush: drain + write, then acknowledge so the
+				// blocked Flush() caller can proceed (used by FatalPanic).
 				f.flush()
+				close(done)
+			case <-f.rotateCh:
+				// Force re-opening the outfile on the next write.
+				// Drained here (not only from write()) so that Rotate()
+				// makes progress even when no log messages arrive.
+				f.lastFileName = ""
 			case <-ctx.Done():
 				f.flush()
-				f.fd.Close()
+				// f.fd is only populated after the first getWriter() call;
+				// guard against a nil pointer when the logger is shut down
+				// before anything has been written.
+				if f.fd != nil {
+					f.fd.Close()
+				}
 				return
 			}
 		}
@@ -100,21 +159,43 @@ func (f *file) RawWithColors(now time.Time, message, coloredMessage string) {
 	panic("Colors not supported in file logger")
 }
 
-func (f *file) Pause()  { f.pauseCh <- struct{}{} }
-func (f *file) Resume() { f.resumeCh <- struct{}{} }
-func (f *file) Flush()  { f.flushCh <- struct{}{} }
+// signal performs a non-blocking, coalescing send on a capacity-1 control
+// channel. If a signal is already pending the new one is dropped, which is
+// the desired behaviour for idempotent operations such as Pause/Rotate/Flush.
+func signal(ch chan struct{}) {
+	select {
+	case ch <- struct{}{}:
+	default:
+	}
+}
 
-func (f *file) Rotate()            { f.rotateCh <- struct{}{} }
-func (*file) SupportsColors() bool { return false }
+func (f *file) Pause()  { signal(f.pauseCh) }
+func (f *file) Resume() { signal(f.resumeCh) }
+func (f *file) Rotate() { signal(f.rotateCh) }
 
-func (f *file) write(m *fileMessageBuf) {
+// Flush synchronously drains any queued messages and writes the bufio buffer to
+// disk, blocking until the logger goroutine acknowledges. The crash path
+// (dlog.FatalPanic) depends on this: with an async signal the process could
+// panic and unwind before the goroutine drained, losing buffered diagnostics.
+// A bounded timeout guards against a deadlock when the goroutine is paused or
+// has already exited (Flush racing shutdown), in which case the ctx.Done path
+// has already flushed or will flush.
+func (f *file) Flush() {
+	done := make(chan struct{})
 	select {
-	case <-f.rotateCh:
-		// Force re-opening the outfile next time in getWriter.
-		f.lastFileName = ""
-	default:
+	case f.flushCh <- done:
+	case <-time.After(fileFlushTimeout):
+		return
+	}
+	select {
+	case <-done:
+	case <-time.After(fileFlushTimeout):
 	}
+}
+
+func (*file) SupportsColors() bool { return false }
 
+func (f *file) write(m *fileMessageBuf) {
 	var writer *bufio.Writer
 	if f.strategy.Rotation == DailyRotation {
 		writer = f.getWriter(m.now.Format("20060102"))
@@ -150,9 +231,12 @@ func (f *file) getWriter(name string) *bufio.Writer {
 		f.writer.Flush()
 		f.fd.Close()
 	}
-	// Set new writer.
+	// Set new writer. Use a real buffer (fileWriterBufSize) so bulk payload
+	// batches into few write syscalls instead of one-or-two per line. The
+	// logger goroutine's idle ticker and the ctx.Done/flush/pause paths keep
+	// low-volume and shutdown output from being stuck in the buffer.
 	f.fd = newFd
-	f.writer = bufio.NewWriterSize(f.fd, 1)
+	f.writer = bufio.NewWriterSize(f.fd, fileWriterBufSize)
 	f.lastFileName = name
 
 	return f.writer