// Cthulhu screen reader. // Copyright (c) 2026 Storm Dragon // Authors: Storm Dragon, Angus Kola // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // This program is distributed in the hope that it will be useful, but WITHOUT // ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or // FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for // more details. // // You should have received a copy of the GNU General Public License along with // this program. If not, see . package logging import ( "context" "errors" "fmt" "io" "log/slog" "os" "sync" "sync/atomic" "time" ) const ( defaultDiagnosticsCapacity = 256 defaultStopTimeout = 2 * time.Second defaultCancelGrace = 100 * time.Millisecond ) type DiagnosticsWriter interface { Write([]byte) (int, error) Cancel() error Close() error } type AsyncOptions struct { Options Capacity int StopTimeout time.Duration CancelGrace time.Duration } type diagnosticRecord struct { data []byte critical bool } type Diagnostics struct { writer DiagnosticsWriter formatter *Handler logger *slog.Logger queue []diagnosticRecord wake chan struct{} stopSignal chan struct{} done chan struct{} mutex sync.Mutex stopOnce sync.Once errorsMutex sync.Mutex stopMutex sync.Mutex writeErrors []error stopError error stopSet bool dropped atomic.Uint64 reported atomic.Uint64 capacity int stopTimeout time.Duration cancelGrace time.Duration } func NewAsync(writer DiagnosticsWriter, options AsyncOptions) *Diagnostics { if writer == nil { writer = NewPassthroughWriter(nil) } if options.Capacity <= 0 { options.Capacity = defaultDiagnosticsCapacity } if options.StopTimeout <= 0 { options.StopTimeout = defaultStopTimeout } if options.CancelGrace <= 0 { options.CancelGrace = defaultCancelGrace } diagnostics := &Diagnostics{ writer: writer, formatter: newHandler(nil, options.Options), queue: make([]diagnosticRecord, 0, options.Capacity), wake: make(chan struct{}, 1), stopSignal: make(chan struct{}), done: make(chan struct{}), capacity: options.Capacity, stopTimeout: options.StopTimeout, cancelGrace: options.CancelGrace, } diagnostics.logger = slog.New(&asyncHandler{ diagnostics: diagnostics, base: diagnostics.formatter, }) go diagnostics.run() return diagnostics } func (diagnostics *Diagnostics) Logger() *slog.Logger { return diagnostics.logger } func (diagnostics *Diagnostics) Dropped() uint64 { return diagnostics.dropped.Load() } func (diagnostics *Diagnostics) Stop(ctx context.Context) error { if ctx == nil { return errors.New("diagnostics stop requires a non-nil context") } diagnostics.signalStop() stopContext, cancel := context.WithTimeout(ctx, diagnostics.stopTimeout) defer cancel() if !diagnostics.waitOrCancel(stopContext) { diagnostics.setStopError(fmt.Errorf("flush diagnostics: %w", stopContext.Err())) return diagnostics.stopErrorValue() } diagnostics.setStopError(errors.Join(diagnostics.writeErrorsValue()...)) return diagnostics.stopErrorValue() } func (diagnostics *Diagnostics) signalStop() { diagnostics.stopOnce.Do(func() { close(diagnostics.stopSignal) diagnostics.wakeup() }) } func (diagnostics *Diagnostics) waitOrCancel(ctx context.Context) bool { deadline, hasDeadline := ctx.Deadline() if hasDeadline { cancelDelay := time.Until(deadline) - diagnostics.cancelGrace if cancelDelay > 0 { timer := time.NewTimer(cancelDelay) select { case <-diagnostics.done: timer.Stop() return true case <-timer.C: } } } else { select { case <-diagnostics.done: return true case <-time.After(diagnostics.cancelGrace): } } if err := diagnostics.writer.Cancel(); err != nil { diagnostics.appendWriteError(fmt.Errorf("cancel diagnostics writer: %w", err)) } select { case <-diagnostics.done: return true case <-ctx.Done(): return false } } func (diagnostics *Diagnostics) run() { defer close(diagnostics.done) for { diagnostics.mutex.Lock() if len(diagnostics.queue) == 0 { if !diagnostics.accepting() { diagnostics.mutex.Unlock() diagnostics.reportDrops() diagnostics.mutex.Lock() if len(diagnostics.queue) == 0 { diagnostics.mutex.Unlock() break } } else { wake := diagnostics.wake diagnostics.mutex.Unlock() select { case <-wake: case <-diagnostics.stopSignal: } continue } } record := diagnostics.queue[0] diagnostics.queue = diagnostics.queue[1:] diagnostics.mutex.Unlock() if _, err := diagnostics.writer.Write(record.data); err != nil { diagnostics.appendWriteError(fmt.Errorf("write diagnostics: %w", err)) } diagnostics.reportDrops() } if err := diagnostics.writer.Close(); err != nil { diagnostics.appendWriteError(fmt.Errorf("close diagnostics writer: %w", err)) } } func (diagnostics *Diagnostics) accepting() bool { select { case <-diagnostics.stopSignal: return false default: return true } } func (diagnostics *Diagnostics) enqueue(record diagnosticRecord) { diagnostics.mutex.Lock() if !diagnostics.accepting() { diagnostics.mutex.Unlock() diagnostics.dropped.Add(1) return } if len(diagnostics.queue) == diagnostics.capacity { diagnostics.dropped.Add(1) var admitted bool diagnostics.queue, admitted = makeRoom(diagnostics.queue, record.critical) if !admitted { diagnostics.mutex.Unlock() return } } diagnostics.queue = append(diagnostics.queue, record) diagnostics.mutex.Unlock() diagnostics.wakeup() } func makeRoom(queue []diagnosticRecord, incomingCritical bool) ([]diagnosticRecord, bool) { for index, record := range queue { if !record.critical { copy(queue[index:], queue[index+1:]) queue[len(queue)-1] = diagnosticRecord{} return queue[:len(queue)-1], true } } if !incomingCritical { return queue, false } copy(queue, queue[1:]) queue[len(queue)-1] = diagnosticRecord{} return queue[:len(queue)-1], true } func (diagnostics *Diagnostics) reportDrops() { total := diagnostics.dropped.Load() reported := diagnostics.reported.Swap(total) if total == reported { return } record := slog.NewRecord(time.Now(), slog.LevelWarn, "diagnostics queue overflow", 0) record.AddAttrs( slog.Uint64("dropped_messages", total-reported), slog.Uint64("total_dropped_messages", total), ) if _, err := diagnostics.writer.Write(diagnostics.formatter.format(record)); err != nil { diagnostics.appendWriteError(fmt.Errorf("write diagnostics overflow report: %w", err)) } } func (diagnostics *Diagnostics) wakeup() { select { case diagnostics.wake <- struct{}{}: default: } } func (diagnostics *Diagnostics) appendWriteError(err error) { diagnostics.errorsMutex.Lock() defer diagnostics.errorsMutex.Unlock() diagnostics.writeErrors = append(diagnostics.writeErrors, err) } func (diagnostics *Diagnostics) writeErrorsValue() []error { diagnostics.errorsMutex.Lock() defer diagnostics.errorsMutex.Unlock() return append([]error(nil), diagnostics.writeErrors...) } func (diagnostics *Diagnostics) setStopError(err error) { diagnostics.stopMutex.Lock() defer diagnostics.stopMutex.Unlock() if !diagnostics.stopSet { diagnostics.stopError = err diagnostics.stopSet = true } } func (diagnostics *Diagnostics) stopErrorValue() error { diagnostics.stopMutex.Lock() defer diagnostics.stopMutex.Unlock() return diagnostics.stopError } type asyncHandler struct { diagnostics *Diagnostics base *Handler } func (handler *asyncHandler) Enabled(ctx context.Context, level slog.Level) bool { return handler.base.Enabled(ctx, level) } func (handler *asyncHandler) Handle(_ context.Context, record slog.Record) error { if !handler.base.Enabled(context.Background(), record.Level) { return nil } handler.diagnostics.enqueue(diagnosticRecord{ data: handler.base.format(record), critical: record.Level >= slog.LevelWarn, }) return nil } func (handler *asyncHandler) WithAttrs(attrs []slog.Attr) slog.Handler { clone := *handler clone.base = handler.base.WithAttrs(attrs).(*Handler) return &clone } func (handler *asyncHandler) WithGroup(name string) slog.Handler { clone := *handler clone.base = handler.base.WithGroup(name).(*Handler) return &clone } type passthroughWriter struct { writer io.Writer } func NewPassthroughWriter(writer io.Writer) DiagnosticsWriter { if writer == nil { writer = io.Discard } return passthroughWriter{writer: writer} } func (writer passthroughWriter) Write(data []byte) (int, error) { return writer.writer.Write(data) } func (writer passthroughWriter) Cancel() error { return nil } func (writer passthroughWriter) Close() error { return nil } type fileWriter struct { file *os.File fallback io.Writer closeOnce sync.Once } func OpenFile(path string, fallback io.Writer) (DiagnosticsWriter, error) { if path == "" || path == "." || path == ".." { return nil, errors.New("invalid diagnostics file path") } file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600) if err != nil { return nil, fmt.Errorf("open diagnostics file: %w", err) } return &fileWriter{file: file, fallback: fallback}, nil } func (writer *fileWriter) Write(data []byte) (int, error) { if writer.fallback != nil { if _, err := writer.fallback.Write(data); err != nil { return 0, fmt.Errorf("write diagnostics fallback: %w", err) } } return writer.file.Write(data) } func (writer *fileWriter) Cancel() error { return writer.close() } func (writer *fileWriter) Close() error { return writer.close() } func (writer *fileWriter) close() error { var closeError error writer.closeOnce.Do(func() { closeError = writer.file.Close() }) if errors.Is(closeError, os.ErrClosed) { return nil } return closeError }