KamadoPool/api/internal/logmon/tailer.go

// Package logmon tails the ckpool log file looking for notable events —
// primarily block-solve lines, which are the most reliable signal we have
// that a block was found, short of a ZMQ hashblock subscription.
//
// ckpool-solo logs a line like:
//
//	Solved and confirmed block 840123
//
// from stratifier.c via LOGWARNING when a submitted share passes network
// difficulty and bitcoind confirms acceptance. We parse these lines,
// emit BlockEvent values on Events, and let the aggregator enrich them
// with hash/reward via bitcoind RPC.
package logmon

import (
	"bufio"
	"context"
	"errors"
	"io"
	"log/slog"
	"os"
	"regexp"
	"strconv"
	"time"
)

// BlockEvent is emitted when the tailer sees a "Solved and confirmed block"
// line in the ckpool log. Hash and worker are populated later by the
// aggregator once it cross-references bitcoind. ShareDiff is the
// difficulty of the winning share, captured from the "Possible block
// solve" line that precedes the confirmation line.
type BlockEvent struct {
	Height    int64     `json:"height"`
	SeenAt    time.Time `json:"seen_at"`
	RawLine   string    `json:"raw_line"`
	ShareDiff float64   `json:"share_diff,omitempty"`
}

// Tailer follows a log file, surviving rotation/truncation, and emits
// parsed events. Create with New, then Run in a goroutine.
type Tailer struct {
	Path     string
	Events   chan BlockEvent
	Log      *slog.Logger
	PollWait time.Duration // how long to sleep between EOF polls

	// lastSolveDiff remembers the share diff from the most recent
	// "Possible block solve" line so handleLine can attach it to the
	// subsequent "Solved and confirmed block" event. Reset after use.
	lastSolveDiff float64
}

func New(path string, log *slog.Logger) *Tailer {
	return &Tailer{
		Path:     path,
		Events:   make(chan BlockEvent, 16),
		Log:      log,
		PollWait: 500 * time.Millisecond,
	}
}

var (
	solvedRE = regexp.MustCompile(`Solved and confirmed block\s+(\d+)`)
	// Matches the three "Possible ... block solve ... diff <float>" lines
	// ckpool emits from stratifier.c right before a block is submitted:
	//   "Possible block solve diff N !"
	//   "Possible stale share block solve diff N !"
	//   "Submitting possible block solve share diff N !"
	//   "Possible remote block solve diff N !"
	solveDiffRE = regexp.MustCompile(`(?:Possible|Submitting[^"]*possible).*block solve.*diff\s+([0-9eE.+-]+)`)
)

// Run blocks until ctx is cancelled. It opens the file, seeks to the end,
// and reads new lines as they are appended. If the file is rotated
// (shrinks, or inode changes), it re-opens.
func (t *Tailer) Run(ctx context.Context) {
	defer close(t.Events)

	var (
		f       *os.File
		reader  *bufio.Reader
		lastIno uint64
		lastPos int64
	)

	open := func() error {
		if f != nil {
			_ = f.Close()
		}
		nf, err := os.Open(t.Path)
		if err != nil {
			return err
		}
		// Start at end on first open so we don't replay old events.
		if _, err := nf.Seek(0, io.SeekEnd); err != nil {
			_ = nf.Close()
			return err
		}
		st, err := nf.Stat()
		if err != nil {
			_ = nf.Close()
			return err
		}
		f = nf
		reader = bufio.NewReader(f)
		lastIno = inodeOf(st)
		lastPos, _ = f.Seek(0, io.SeekCurrent)
		return nil
	}

	// Initial open; retry on failure until the file exists.
	for {
		if err := open(); err != nil {
			t.Log.Warn("logmon: waiting for log file", "path", t.Path, "err", err)
			if !sleep(ctx, 2*time.Second) {
				return
			}
			continue
		}
		break
	}
	defer func() {
		if f != nil {
			_ = f.Close()
		}
	}()

	for {
		if ctx.Err() != nil {
			return
		}
		line, err := reader.ReadString('\n')
		if len(line) > 0 {
			t.handleLine(line)
			lastPos, _ = f.Seek(0, io.SeekCurrent)
		}
		if err == nil {
			continue
		}
		if !errors.Is(err, io.EOF) {
			t.Log.Warn("logmon: read error, reopening", "err", err)
			if !sleep(ctx, t.PollWait) {
				return
			}
			_ = open()
			continue
		}

		// EOF: check for rotation (inode changed) or truncation (size < pos).
		if st, statErr := os.Stat(t.Path); statErr == nil {
			if inodeOf(st) != lastIno || st.Size() < lastPos {
				t.Log.Info("logmon: log rotated, reopening", "path", t.Path)
				if err := open(); err != nil {
					t.Log.Warn("logmon: reopen failed", "err", err)
				}
				continue
			}
		}
		if !sleep(ctx, t.PollWait) {
			return
		}
	}
}

func (t *Tailer) handleLine(line string) {
	if m := solveDiffRE.FindStringSubmatch(line); m != nil {
		if d, err := strconv.ParseFloat(m[1], 64); err == nil {
			t.lastSolveDiff = d
		}
		return
	}
	m := solvedRE.FindStringSubmatch(line)
	if m == nil {
		return
	}
	height, err := strconv.ParseInt(m[1], 10, 64)
	if err != nil {
		return
	}
	ev := BlockEvent{
		Height:    height,
		SeenAt:    time.Now(),
		RawLine:   line,
		ShareDiff: t.lastSolveDiff,
	}
	t.lastSolveDiff = 0
	select {
	case t.Events <- ev:
		t.Log.Info("logmon: block solved", "height", height, "share_diff", ev.ShareDiff)
	default:
		t.Log.Warn("logmon: events channel full, dropping", "height", height)
	}
}

// sleep returns false if ctx was cancelled during the wait.
func sleep(ctx context.Context, d time.Duration) bool {
	select {
	case <-ctx.Done():
		return false
	case <-time.After(d):
		return true
	}
}