package distworker import ( "sync" "time" "rocketgit.ru/rsmon/worker/internal/wire" ) // recentResultsSize is the in-memory ring buffer capacity for the // last N check results produced by the worker. The webapp reads // from this buffer for the /checks page and the recent-results // counters on /overview. const recentResultsSize = 200 // recentNotificationsSize mirrors recentResultsSize for emitted // notifications. const recentNotificationsSize = 100 // ResultRow is one row from the worker's in-memory result ring // buffer. Kept in the distworker package so the webapp can read it // without going through wire (which is a payload envelope, not a // stable render type). type ResultRow struct { MonitorID int64 CheckID int64 Kind string Host string State string DurationMs int64 Error string At time.Time } // NotificationRow is one row from the worker's notification ring buffer. // Delegated rows use only JobID, Method, Status, DurationMs, and At. They // deliberately omit delivery inputs and provider output. type NotificationRow struct { Kind string // "email", "telegram_private", "telegram_group" Channel string Subject string Body string OK bool Error string At time.Time JobID string Method string Status string DurationMs int } // resultBuffer is a thread-safe FIFO ring buffer of ResultRow. The // Runner owns one; both the dispatcher goroutine and the webapp // readers can touch it concurrently. type resultBuffer struct { mu sync.RWMutex buf []ResultRow head int size int } func newResultBuffer() *resultBuffer { return &resultBuffer{buf: make([]ResultRow, recentResultsSize)} } // add appends one entry, evicting the oldest when full. func (b *resultBuffer) add(r *ResultRow) { if b == nil { return } b.mu.Lock() defer b.mu.Unlock() b.buf[b.head] = *r b.head = (b.head + 1) % len(b.buf) if b.size < len(b.buf) { b.size++ } } // snapshot returns the most recent n rows in chronological order. // n <= 0 returns an empty slice. func (b *resultBuffer) snapshot(n int) []ResultRow { return ringSnapshot(&b.mu, &b.head, &b.size, &b.buf, n) } // notificationBuffer mirrors resultBuffer for NotificationRow. type notificationBuffer struct { mu sync.RWMutex buf []NotificationRow head int size int } func newNotificationBuffer() *notificationBuffer { return ¬ificationBuffer{buf: make([]NotificationRow, recentNotificationsSize)} } func (b *notificationBuffer) add(r *NotificationRow) { if b == nil { return } b.mu.Lock() defer b.mu.Unlock() b.buf[b.head] = *r b.head = (b.head + 1) % len(b.buf) if b.size < len(b.buf) { b.size++ } } func (b *notificationBuffer) snapshot(n int) []NotificationRow { return ringSnapshot(&b.mu, &b.head, &b.size, &b.buf, n) } // ringSnapshot is a generic FIFO ring-buffer snapshot. The caller // passes the mutex/head/size/buf by pointer; the lock is taken // while reading. n <= 0 returns nil. The returned slice is a copy // so callers can hand it to non-locking code paths (e.g. the // webapp's templates) without holding the buffer lock. func ringSnapshot[T any](mu *sync.RWMutex, head, size *int, buf *[]T, n int) []T { if mu == nil || n <= 0 { return nil } mu.RLock() defer mu.RUnlock() if *size == 0 { return nil } if n > *size { n = *size } out := make([]T, 0, n) start := (*head - n + len(*buf)) % len(*buf) for i := 0; i < n; i++ { idx := (start + i) % len(*buf) out = append(out, (*buf)[idx]) } return out } // resultRowFromReport converts a wire.CheckResultReport into the // internal ResultRow type the webapp renders. func resultRowFromReport(env *resultEnvelope, report *wire.CheckResultReport, at time.Time) *ResultRow { row := &ResultRow{ MonitorID: report.MonitorID, CheckID: report.CheckID, Kind: env.job.Kind, Host: env.job.Host, State: report.State, DurationMs: report.DurationMs, At: at, } if report.Error != nil { row.Error = *report.Error } return row }