Make ckpool client resilient to warm-up EOFs

The stratifier can accept a connection and close it without
writing a response during early startup — the refresh loop
then logs 'ckpool poolstats failed: read len: EOF' on every
tick until the stratifier is ready, which is noisy and looks
like a real fault.

- client.Send retries once after 200ms if the server closed
  the connection before any bytes were read (io.EOF anywhere
  in the wrapped chain).
- Aggregator tracks a ckFailStreak counter: first two
  consecutive failures log at DEBUG, third and beyond escalate
  to WARN. Successful refreshes reset the streak.
This commit is contained in:
satoshi
2026-04-14 10:58:26 +03:00
parent 82941939e5
commit c92a991e89
3 changed files with 34 additions and 2 deletions
+17
View File
@@ -51,7 +51,24 @@ var ErrProtocol = errors.New("ckpool: protocol error")
// Send delivers a single command to the named ckpool socket and returns the // Send delivers a single command to the named ckpool socket and returns the
// raw response bytes. `sockName` is one of SocketListener, SocketStratifier, // raw response bytes. `sockName` is one of SocketListener, SocketStratifier,
// or SocketConnector. // or SocketConnector.
//
// If the server closes the connection before writing a response (common
// during ckpool warm-up — the stratifier accepts the socket but its stats
// aren't populated yet), Send retries once after a short backoff.
func (c *Client) Send(ctx context.Context, sockName, cmd string) ([]byte, error) { func (c *Client) Send(ctx context.Context, sockName, cmd string) ([]byte, error) {
buf, err := c.sendOnce(ctx, sockName, cmd)
if err != nil && errors.Is(err, io.EOF) {
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-time.After(200 * time.Millisecond):
}
return c.sendOnce(ctx, sockName, cmd)
}
return buf, err
}
func (c *Client) sendOnce(ctx context.Context, sockName, cmd string) ([]byte, error) {
if cmd == "" { if cmd == "" {
return nil, fmt.Errorf("ckpool: empty command") return nil, fmt.Errorf("ckpool: empty command")
} }
+16 -1
View File
@@ -59,6 +59,12 @@ type Aggregator struct {
mu sync.RWMutex mu sync.RWMutex
snap Snapshot snap Snapshot
blocks []BlockRecord blocks []BlockRecord
// ckFailStreak counts consecutive refreshes where CKPool returned an
// error. The first failure after a streak of successes is logged at
// DEBUG (likely a transient warm-up or lock hiccup); repeated failures
// escalate to WARN.
ckFailStreak int
} }
func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator { func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator {
@@ -108,8 +114,17 @@ func (a *Aggregator) refresh(ctx context.Context) {
next.HashrateHs1h = ckpool.DSPSToHashrate(ps.DSPS60) next.HashrateHs1h = ckpool.DSPSToHashrate(ps.DSPS60)
next.HashrateHs24h = ckpool.DSPSToHashrate(ps.DSPS1440) next.HashrateHs24h = ckpool.DSPSToHashrate(ps.DSPS1440)
next.CKPoolOK = true next.CKPoolOK = true
a.ckFailStreak = 0
} else { } else {
a.Log.Warn("ckpool poolstats failed", "err", err) a.ckFailStreak++
// First failure or two: likely transient (stratifier warm-up or
// a lock hiccup during a reconnect). Only escalate to WARN after
// three consecutive failures.
if a.ckFailStreak >= 3 {
a.Log.Warn("ckpool poolstats failed", "err", err, "streak", a.ckFailStreak)
} else {
a.Log.Debug("ckpool poolstats transient error", "err", err, "streak", a.ckFailStreak)
}
next.LastError = err.Error() next.LastError = err.Error()
} }
if u, err := a.CK.Uptime(ctx); err == nil { if u, err := a.CK.Uptime(ctx); err == nil {
+1 -1
View File
@@ -16,6 +16,6 @@
"svelte": "^5.0.0", "svelte": "^5.0.0",
"svelte-check": "^4.0.0", "svelte-check": "^4.0.0",
"typescript": "^5.6.0", "typescript": "^5.6.0",
"vite": "^5.4.0" "vite": "^6.0.0"
} }
} }