metrics: attribute hashrate to build, device, kernel and job outcomes
ci / fmt (pull_request) Successful in 20s
bench / build (pull_request) Successful in 1m2s
ci / clippy (pull_request) Successful in 1m37s
ci / doc (pull_request) Successful in 1m58s
bench / measure (pull_request) Successful in 2m55s
ci / test (pull_request) Successful in 6m50s
ci / fmt (pull_request) Successful in 20s
bench / build (pull_request) Successful in 1m2s
ci / clippy (pull_request) Successful in 1m37s
ci / doc (pull_request) Successful in 1m58s
bench / measure (pull_request) Successful in 2m55s
ci / test (pull_request) Successful in 6m50s
quantus/miner#9. Additive metrics in a lair-owned module of the metrics
crate; origin's metrics are untouched so the fleet dashboard keeps working
across origin merges.
Identity:
- miner_build_info{version, commit}: the join key for everything below.
- miner_config_info{engine, gpu_batch_size, gpu_devices, cpu_workers,
gpu_throttle_ms}: two deploys of one commit with different flags are
different experiments.
Per device (handles resolved once at engine init, no label lookups in the
batch path):
- miner_device_hashes_total{device, kernel}, miner_device_solutions_total,
miner_device_lost_total.
- miner_gpu_batch_seconds{device, kernel, phase=gpu|host}: submit-to-mapped
on the device versus everything else in the batch. This is the bubble
#4, #5 and #6 attack, measured directly.
Jobs and results:
- miner_jobs_received_total, miner_stale_hashes_total{engine},
miner_job_pickup_seconds{engine} (issued to picked up; the cancel latency
of a busy worker), miner_results_submitted_total,
miner_results_send_failed_total, miner_seal_latency_seconds (found to
sent), miner_job_idle_seconds_total (sent to next job, node-attributable).
Connection: miner_connects_total, miner_connect_failures_total,
miner_disconnects_total, miner_connected, miner_disconnected_seconds_total.
Origin-owned code touched, each block marked lair: engine-gpu gains a
metrics dependency, a DeviceMetrics handle on GpuContext and timing points
in run_single_batch; miner-service gains found_at on WorkerResult,
created_at on MiningJob and the recording calls; miner-cli sets build info
at startup. Deploy validate now asserts miner_build_info carries the
deployed commit.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CBgs2nSi4H2mdh8kD8vMX5
This commit is contained in:
@@ -307,6 +307,13 @@ jobs:
|
||||
else
|
||||
echo " ok connected; miner_gpu_devices ${devs%.*}"
|
||||
fi
|
||||
# The build-info gauge is what ties every other metric to a commit
|
||||
# (quantus/miner#9). In deploy mode it must carry this commit.
|
||||
if grep -q "^miner_build_info{.*commit=\"${{ github.sha }}\"" <<<"$m"; then
|
||||
echo " ok miner_build_info carries ${{ github.sha }}"
|
||||
elif [ "${{ github.event.inputs.mode }}" != validate ]; then
|
||||
echo " miner_build_info does not carry commit ${{ github.sha }}" >&2; fail=1
|
||||
fi
|
||||
|
||||
deadline=$((SECONDS + 120))
|
||||
h1=""; h2=""; ok=0
|
||||
|
||||
Generated
+1
@@ -613,6 +613,7 @@ dependencies = [
|
||||
"futures",
|
||||
"hex",
|
||||
"log",
|
||||
"metrics",
|
||||
"pow-core",
|
||||
"primitive-types 0.13.1",
|
||||
"qp-plonky2",
|
||||
|
||||
@@ -19,6 +19,7 @@ simd-poseidon2 = []
|
||||
|
||||
[dependencies]
|
||||
engine-cpu = { path = "../engine-cpu" }
|
||||
metrics = { path = "../metrics" } # lair: per-device metrics (quantus/miner#9)
|
||||
pow-core = { path = "../pow-core" }
|
||||
primitive-types = { workspace = true }
|
||||
log = { workspace = true }
|
||||
|
||||
@@ -20,6 +20,8 @@ use std::sync::{
|
||||
|
||||
/// Represents a single GPU device context.
|
||||
struct GpuContext {
|
||||
// lair: per-device metric handles, labelled with the kernel id (quantus/miner#9).
|
||||
metrics: metrics::DeviceMetrics,
|
||||
device: wgpu::Device,
|
||||
queue: wgpu::Queue,
|
||||
pipeline: wgpu::ComputePipeline,
|
||||
@@ -443,6 +445,7 @@ impl GpuEngine {
|
||||
|
||||
initialized.push(InitializedGpu {
|
||||
context: Arc::new(GpuContext {
|
||||
metrics: metrics::DeviceMetrics::new(i, kernel.id()),
|
||||
device,
|
||||
queue,
|
||||
pipeline,
|
||||
@@ -698,8 +701,9 @@ impl MinerEngine for GpuEngine {
|
||||
total_hashes += hash_count;
|
||||
}
|
||||
BatchResult::DeviceLost => {
|
||||
// GPU device is lost/unresponsive - mark as permanently dead
|
||||
// and clear resources to prevent "buffer already mapped" panics
|
||||
gpu_ctx.metrics.record_device_lost(); // lair
|
||||
// GPU device is lost/unresponsive - mark as permanently dead
|
||||
// and clear resources to prevent "buffer already mapped" panics
|
||||
DEVICE_LOST.with(|lost| *lost.borrow_mut() = Some(self.engine_id));
|
||||
WORKER_RESOURCES.with(|res| *res.borrow_mut() = None);
|
||||
|
||||
@@ -793,7 +797,8 @@ fn run_single_batch(
|
||||
batch_start: U512,
|
||||
batch_size: u32,
|
||||
) -> BatchResult {
|
||||
// Calculate dispatch configuration for this batch
|
||||
let batch_start_at = std::time::Instant::now(); // lair: batch phase timing
|
||||
// Calculate dispatch configuration for this batch
|
||||
let threads_per_workgroup = 256u32;
|
||||
let limits = gpu_ctx.device.limits();
|
||||
let max_workgroups = limits.max_compute_workgroups_per_dimension;
|
||||
@@ -865,6 +870,7 @@ fn run_single_batch(
|
||||
);
|
||||
|
||||
gpu_ctx.queue.submit(Some(encoder.finish()));
|
||||
let submitted_at = std::time::Instant::now(); // lair
|
||||
|
||||
// Wait for GPU to complete (blocking)
|
||||
let buffer_slice = resources.staging_buffer.slice(..);
|
||||
@@ -911,6 +917,12 @@ fn run_single_batch(
|
||||
|
||||
// Only reach here if final_status == 1 (success), buffer is mapped
|
||||
debug_assert_eq!(final_status, 1);
|
||||
// lair: gpu = submit to mapped; host = the rest of the batch so far plus
|
||||
// the readback below, which is small and constant.
|
||||
let gpu_time = submitted_at.elapsed();
|
||||
gpu_ctx
|
||||
.metrics
|
||||
.observe_batch(gpu_time, batch_start_at.elapsed() - gpu_time);
|
||||
|
||||
// Read results
|
||||
let data = buffer_slice.get_mapped_range();
|
||||
@@ -939,6 +951,8 @@ fn run_single_batch(
|
||||
drop(data);
|
||||
resources.staging_buffer.unmap();
|
||||
|
||||
gpu_ctx.metrics.record_hashes(hashes_computed); // lair
|
||||
gpu_ctx.metrics.record_solution(); // lair
|
||||
return BatchResult::Found {
|
||||
candidate: Candidate { nonce, work, hash },
|
||||
hash_count: hashes_computed,
|
||||
@@ -948,6 +962,7 @@ fn run_single_batch(
|
||||
drop(data);
|
||||
resources.staging_buffer.unmap();
|
||||
|
||||
gpu_ctx.metrics.record_hashes(dispatched_nonces); // lair
|
||||
BatchResult::NotFound {
|
||||
hash_count: dispatched_nonces,
|
||||
}
|
||||
|
||||
@@ -0,0 +1,340 @@
|
||||
//! lair: metrics that attribute performance to builds, devices and jobs
|
||||
//! (quantus/miner#9). Additive to origin's metrics in `lib.rs`; nothing there
|
||||
//! changes, so the fleet dashboard keeps working across origin merges.
|
||||
//!
|
||||
//! Naming: `miner_*` like origin's. Per-device series carry `device` (engine
|
||||
//! index) and `kernel` (the kernel id the engine selected), so a silent
|
||||
//! fallback from one kernel to another shows up as a label change rather than
|
||||
//! as an unexplained hashrate drop.
|
||||
|
||||
use crate::REGISTRY;
|
||||
use once_cell::sync::Lazy;
|
||||
use prometheus::{
|
||||
Counter, Histogram, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge,
|
||||
IntGaugeVec, Opts,
|
||||
};
|
||||
use std::time::Duration;
|
||||
|
||||
fn reg<M: prometheus::core::Collector + Clone + 'static>(m: M) -> M {
|
||||
REGISTRY
|
||||
.register(Box::new(m.clone()))
|
||||
.expect("register lair metric");
|
||||
m
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Identity: which build and which configuration produced the numbers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
static BUILD_INFO: Lazy<IntGaugeVec> = Lazy::new(|| {
|
||||
reg(IntGaugeVec::new(
|
||||
Opts::new(
|
||||
"miner_build_info",
|
||||
"Build identity of the running miner, always 1",
|
||||
),
|
||||
&["version", "commit"],
|
||||
)
|
||||
.expect("miner_build_info"))
|
||||
});
|
||||
|
||||
static CONFIG_INFO: Lazy<IntGaugeVec> = Lazy::new(|| {
|
||||
reg(IntGaugeVec::new(
|
||||
Opts::new(
|
||||
"miner_config_info",
|
||||
"Effective mining configuration, always 1",
|
||||
),
|
||||
&[
|
||||
"engine",
|
||||
"gpu_batch_size",
|
||||
"gpu_devices",
|
||||
"cpu_workers",
|
||||
"gpu_throttle_ms",
|
||||
],
|
||||
)
|
||||
.expect("miner_config_info"))
|
||||
});
|
||||
|
||||
/// Set once at startup. `commit` is the SHA embedded by build.rs.
|
||||
pub fn set_build_info(version: &str, commit: &str) {
|
||||
BUILD_INFO.with_label_values(&[version, commit]).set(1);
|
||||
}
|
||||
|
||||
/// Set once the engines are resolved. Two deploys of one commit with different
|
||||
/// flags must not look like the same experiment.
|
||||
pub fn set_config_info(
|
||||
engine: &str,
|
||||
gpu_batch_size: u32,
|
||||
gpu_devices: usize,
|
||||
cpu_workers: usize,
|
||||
gpu_throttle_ms: u64,
|
||||
) {
|
||||
CONFIG_INFO
|
||||
.with_label_values(&[
|
||||
engine,
|
||||
&gpu_batch_size.to_string(),
|
||||
&gpu_devices.to_string(),
|
||||
&cpu_workers.to_string(),
|
||||
&gpu_throttle_ms.to_string(),
|
||||
])
|
||||
.set(1);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Per device: throughput, outcomes, and where the time goes inside a batch
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
static DEVICE_HASHES: Lazy<IntCounterVec> = Lazy::new(|| {
|
||||
reg(IntCounterVec::new(
|
||||
Opts::new(
|
||||
"miner_device_hashes_total",
|
||||
"Hashes computed, per GPU device and kernel",
|
||||
),
|
||||
&["device", "kernel"],
|
||||
)
|
||||
.expect("miner_device_hashes_total"))
|
||||
});
|
||||
|
||||
static DEVICE_SOLUTIONS: Lazy<IntCounterVec> = Lazy::new(|| {
|
||||
reg(IntCounterVec::new(
|
||||
Opts::new(
|
||||
"miner_device_solutions_total",
|
||||
"Solutions found, per GPU device and kernel",
|
||||
),
|
||||
&["device", "kernel"],
|
||||
)
|
||||
.expect("miner_device_solutions_total"))
|
||||
});
|
||||
|
||||
static DEVICE_LOST: Lazy<IntCounterVec> = Lazy::new(|| {
|
||||
reg(IntCounterVec::new(
|
||||
Opts::new(
|
||||
"miner_device_lost_total",
|
||||
"Times a GPU device was lost or unresponsive and its worker stopped",
|
||||
),
|
||||
&["device", "kernel"],
|
||||
)
|
||||
.expect("miner_device_lost_total"))
|
||||
});
|
||||
|
||||
static GPU_BATCH_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
|
||||
reg(HistogramVec::new(
|
||||
HistogramOpts::new(
|
||||
"miner_gpu_batch_seconds",
|
||||
"Per-batch time split into the GPU executing (phase=gpu) and the host \
|
||||
preparing, submitting and reading back (phase=host)",
|
||||
)
|
||||
.buckets(vec![
|
||||
0.0005, 0.001, 0.002, 0.004, 0.008, 0.016, 0.032, 0.064, 0.128, 0.256, 0.512, 1.0, 2.0,
|
||||
]),
|
||||
&["device", "kernel", "phase"],
|
||||
)
|
||||
.expect("miner_gpu_batch_seconds"))
|
||||
});
|
||||
|
||||
/// Handles for one device, resolved once so the per-batch path does no label
|
||||
/// lookups. Created by the engine when it initialises a device.
|
||||
#[derive(Clone)]
|
||||
pub struct DeviceMetrics {
|
||||
hashes: IntCounter,
|
||||
solutions: IntCounter,
|
||||
lost: IntCounter,
|
||||
batch_gpu: Histogram,
|
||||
batch_host: Histogram,
|
||||
}
|
||||
|
||||
impl DeviceMetrics {
|
||||
pub fn new(device: usize, kernel: &str) -> Self {
|
||||
let d = device.to_string();
|
||||
Self {
|
||||
hashes: DEVICE_HASHES.with_label_values(&[&d, kernel]),
|
||||
solutions: DEVICE_SOLUTIONS.with_label_values(&[&d, kernel]),
|
||||
lost: DEVICE_LOST.with_label_values(&[&d, kernel]),
|
||||
batch_gpu: GPU_BATCH_SECONDS.with_label_values(&[&d, kernel, "gpu"]),
|
||||
batch_host: GPU_BATCH_SECONDS.with_label_values(&[&d, kernel, "host"]),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn record_hashes(&self, n: u64) {
|
||||
self.hashes.inc_by(n);
|
||||
}
|
||||
|
||||
pub fn record_solution(&self) {
|
||||
self.solutions.inc();
|
||||
}
|
||||
|
||||
pub fn record_device_lost(&self) {
|
||||
self.lost.inc();
|
||||
}
|
||||
|
||||
/// `gpu` is submit-to-completion on the device; `host` is everything else
|
||||
/// in the batch (buffer writes, encoding, readback, bookkeeping).
|
||||
pub fn observe_batch(&self, gpu: Duration, host: Duration) {
|
||||
self.batch_gpu.observe(gpu.as_secs_f64());
|
||||
self.batch_host.observe(host.as_secs_f64());
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Jobs and results: the efficiency side of every throughput trade-off
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
static JOBS_RECEIVED: Lazy<IntCounter> = Lazy::new(|| {
|
||||
reg(
|
||||
IntCounter::new("miner_jobs_received_total", "Jobs received from the node")
|
||||
.expect("miner_jobs_received_total"),
|
||||
)
|
||||
});
|
||||
|
||||
static STALE_HASHES: Lazy<IntCounterVec> = Lazy::new(|| {
|
||||
reg(IntCounterVec::new(
|
||||
Opts::new(
|
||||
"miner_stale_hashes_total",
|
||||
"Hashes reported for a job after a newer job had been issued (wasted work)",
|
||||
),
|
||||
&["engine"],
|
||||
)
|
||||
.expect("miner_stale_hashes_total"))
|
||||
});
|
||||
|
||||
static JOB_PICKUP_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
|
||||
reg(HistogramVec::new(
|
||||
HistogramOpts::new(
|
||||
"miner_job_pickup_seconds",
|
||||
"Job issued to a worker starting on it. For a busy worker this is the \
|
||||
time to notice cancellation and finish the in-flight batch",
|
||||
)
|
||||
.buckets(vec![
|
||||
0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0,
|
||||
]),
|
||||
&["engine"],
|
||||
)
|
||||
.expect("miner_job_pickup_seconds"))
|
||||
});
|
||||
|
||||
static RESULTS_SUBMITTED: Lazy<IntCounter> = Lazy::new(|| {
|
||||
reg(IntCounter::new(
|
||||
"miner_results_submitted_total",
|
||||
"Solutions sent to the node",
|
||||
)
|
||||
.expect("miner_results_submitted_total"))
|
||||
});
|
||||
|
||||
static RESULTS_SEND_FAILED: Lazy<IntCounter> = Lazy::new(|| {
|
||||
reg(IntCounter::new(
|
||||
"miner_results_send_failed_total",
|
||||
"Solutions that could not be sent to the node",
|
||||
)
|
||||
.expect("miner_results_send_failed_total"))
|
||||
});
|
||||
|
||||
static SEAL_LATENCY: Lazy<Histogram> = Lazy::new(|| {
|
||||
reg(Histogram::with_opts(
|
||||
HistogramOpts::new(
|
||||
"miner_seal_latency_seconds",
|
||||
"Solution found on a worker to the result sent to the node",
|
||||
)
|
||||
.buckets(vec![
|
||||
0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0,
|
||||
]),
|
||||
)
|
||||
.expect("miner_seal_latency_seconds"))
|
||||
});
|
||||
|
||||
static JOB_IDLE_SECONDS: Lazy<Counter> = Lazy::new(|| {
|
||||
reg(Counter::new(
|
||||
"miner_job_idle_seconds_total",
|
||||
"Seconds spent between sending a result and receiving the next job (node-attributable idle)",
|
||||
)
|
||||
.expect("miner_job_idle_seconds_total"))
|
||||
});
|
||||
|
||||
pub fn record_job_received() {
|
||||
JOBS_RECEIVED.inc();
|
||||
}
|
||||
|
||||
pub fn record_stale_hashes(engine: &str, n: u64) {
|
||||
STALE_HASHES.with_label_values(&[engine]).inc_by(n);
|
||||
}
|
||||
|
||||
pub fn observe_job_pickup(engine: &str, d: Duration) {
|
||||
JOB_PICKUP_SECONDS
|
||||
.with_label_values(&[engine])
|
||||
.observe(d.as_secs_f64());
|
||||
}
|
||||
|
||||
pub fn record_result_submitted(found_to_sent: Option<Duration>) {
|
||||
RESULTS_SUBMITTED.inc();
|
||||
if let Some(d) = found_to_sent {
|
||||
SEAL_LATENCY.observe(d.as_secs_f64());
|
||||
}
|
||||
}
|
||||
|
||||
pub fn record_result_send_failed() {
|
||||
RESULTS_SEND_FAILED.inc();
|
||||
}
|
||||
|
||||
pub fn record_job_idle(d: Duration) {
|
||||
JOB_IDLE_SECONDS.inc_by(d.as_secs_f64());
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Connection to the node: is a hashrate drop the miner or the node?
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
static CONNECTS: Lazy<IntCounter> = Lazy::new(|| {
|
||||
reg(IntCounter::new(
|
||||
"miner_connects_total",
|
||||
"Connections established to the node",
|
||||
)
|
||||
.expect("miner_connects_total"))
|
||||
});
|
||||
|
||||
static CONNECT_FAILURES: Lazy<IntCounter> = Lazy::new(|| {
|
||||
reg(IntCounter::new(
|
||||
"miner_connect_failures_total",
|
||||
"Connection attempts to the node that failed",
|
||||
)
|
||||
.expect("miner_connect_failures_total"))
|
||||
});
|
||||
|
||||
static DISCONNECTS: Lazy<IntCounter> = Lazy::new(|| {
|
||||
reg(IntCounter::new(
|
||||
"miner_disconnects_total",
|
||||
"Connections to the node that were lost",
|
||||
)
|
||||
.expect("miner_disconnects_total"))
|
||||
});
|
||||
|
||||
static CONNECTED: Lazy<IntGauge> = Lazy::new(|| {
|
||||
reg(
|
||||
IntGauge::new("miner_connected", "1 while connected to the node, else 0")
|
||||
.expect("miner_connected"),
|
||||
)
|
||||
});
|
||||
|
||||
static DISCONNECTED_SECONDS: Lazy<Counter> = Lazy::new(|| {
|
||||
reg(Counter::new(
|
||||
"miner_disconnected_seconds_total",
|
||||
"Seconds spent without a node connection after the first successful connection",
|
||||
)
|
||||
.expect("miner_disconnected_seconds_total"))
|
||||
});
|
||||
|
||||
pub fn record_connected() {
|
||||
CONNECTS.inc();
|
||||
CONNECTED.set(1);
|
||||
}
|
||||
|
||||
pub fn record_connect_failed() {
|
||||
CONNECT_FAILURES.inc();
|
||||
CONNECTED.set(0);
|
||||
}
|
||||
|
||||
pub fn record_disconnected() {
|
||||
DISCONNECTS.inc();
|
||||
CONNECTED.set(0);
|
||||
}
|
||||
|
||||
pub fn record_disconnected_time(d: Duration) {
|
||||
DISCONNECTED_SECONDS.inc_by(d.as_secs_f64());
|
||||
}
|
||||
@@ -39,6 +39,10 @@ use anyhow::Result;
|
||||
|
||||
static REGISTRY: Lazy<Registry> = Lazy::new(Registry::new);
|
||||
|
||||
// lair: build, device and job metrics (quantus/miner#9); see lair.rs.
|
||||
mod lair;
|
||||
pub use lair::*;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Hash Rate Metrics
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -144,6 +144,8 @@ struct Args {
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
let args = Args::parse();
|
||||
// lair: identify this build in the metrics (quantus/miner#9).
|
||||
metrics::set_build_info(env!("CARGO_PKG_VERSION"), env!("MINER_BUILD_SHA"));
|
||||
|
||||
let Some(command) = args.command else {
|
||||
eprintln!("Error: No command provided. Use 'serve' to start mining (defaults to local node at 127.0.0.1:9833).");
|
||||
|
||||
@@ -62,6 +62,8 @@ pub struct WorkerResult {
|
||||
pub hash_count: u64,
|
||||
/// Whether this worker has finished its range.
|
||||
pub completed: bool,
|
||||
/// lair: when the candidate was found (seal latency, quantus/miner#9).
|
||||
pub found_at: Option<std::time::Instant>,
|
||||
}
|
||||
|
||||
/// A successful mining candidate.
|
||||
@@ -90,6 +92,8 @@ pub struct MiningJob {
|
||||
pub ctx: pow_core::JobContext,
|
||||
/// Job ID to detect stale results after job transitions
|
||||
pub job_id: u64,
|
||||
/// lair: when the job was issued (pickup latency, quantus/miner#9).
|
||||
created_at: std::time::Instant,
|
||||
}
|
||||
|
||||
/// Persistent worker thread pool that keeps threads alive between jobs.
|
||||
@@ -208,6 +212,7 @@ impl WorkerPool {
|
||||
let job = MiningJob {
|
||||
ctx,
|
||||
job_id: new_job_id,
|
||||
created_at: std::time::Instant::now(), // lair
|
||||
};
|
||||
|
||||
// Dispatch job to all workers using bounded channels (capacity 16).
|
||||
@@ -340,6 +345,9 @@ fn worker_loop(
|
||||
log::debug!("[WORKER {type_str}-{thread_id}] Drained {skipped} stale jobs from queue");
|
||||
}
|
||||
|
||||
// lair: issued-to-picked-up; for a busy worker this is the cancel latency.
|
||||
metrics::observe_job_pickup(type_str, job.created_at.elapsed());
|
||||
|
||||
// Capture the job's ID for later validation
|
||||
let job_id = job.job_id;
|
||||
log::debug!("[WORKER {type_str}-{thread_id}] Received job {job_id}");
|
||||
@@ -398,6 +406,7 @@ fn worker_loop(
|
||||
candidate: None, // Discard the stale candidate
|
||||
hash_count,
|
||||
completed: true,
|
||||
found_at: None,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
@@ -442,6 +451,7 @@ fn worker_loop(
|
||||
candidate: None,
|
||||
hash_count,
|
||||
completed: true,
|
||||
found_at: None,
|
||||
});
|
||||
break; // Exit the worker loop
|
||||
}
|
||||
@@ -452,6 +462,7 @@ fn worker_loop(
|
||||
};
|
||||
|
||||
// Send result (non-blocking to avoid deadlock if receiver is full)
|
||||
let found_at = candidate.as_ref().map(|_| std::time::Instant::now()); // lair
|
||||
let _ = result_tx.try_send(WorkerResult {
|
||||
thread_id,
|
||||
engine_type,
|
||||
@@ -459,6 +470,7 @@ fn worker_loop(
|
||||
candidate,
|
||||
hash_count,
|
||||
completed: true,
|
||||
found_at,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -562,6 +574,14 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> {
|
||||
cpu_workers,
|
||||
gpu_devices
|
||||
);
|
||||
// lair: expose the effective configuration as labels (quantus/miner#9).
|
||||
metrics::set_config_info(
|
||||
gpu_engine.as_ref().map(|e| e.name()).unwrap_or("cpu"),
|
||||
config.gpu_batch_size,
|
||||
gpu_devices,
|
||||
cpu_workers,
|
||||
config.gpu_throttle_ms,
|
||||
);
|
||||
|
||||
if let Some(ref engine) = cpu_engine {
|
||||
let name = engine.name();
|
||||
|
||||
@@ -38,6 +38,8 @@ pub async fn connect_and_mine(
|
||||
|
||||
let mut reconnect_delay = Duration::from_secs(1);
|
||||
const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(30);
|
||||
// lair: time without a connection, counted after the first success.
|
||||
let mut disconnected_since: Option<Instant> = None;
|
||||
|
||||
loop {
|
||||
log::info!("⛏️ Connecting to node at {}...", node_addr);
|
||||
@@ -45,6 +47,10 @@ pub async fn connect_and_mine(
|
||||
match establish_connection(node_addr, auth_token, tls_cert_sha256).await {
|
||||
Ok((connection, send, recv)) => {
|
||||
log::info!("⛏️ Connected to node at {}", node_addr);
|
||||
metrics::record_connected(); // lair
|
||||
if let Some(t) = disconnected_since.take() {
|
||||
metrics::record_disconnected_time(t.elapsed());
|
||||
}
|
||||
|
||||
let mut authenticated = false;
|
||||
if let Err(e) =
|
||||
@@ -61,6 +67,8 @@ pub async fn connect_and_mine(
|
||||
}
|
||||
log::info!("⛏️ Connection lost: {}", e);
|
||||
}
|
||||
metrics::record_disconnected(); // lair
|
||||
disconnected_since = Some(Instant::now());
|
||||
// Only clear backoff after the node accepted auth (first NewJob).
|
||||
// connect() already treats explicit "auth failed" as permanent;
|
||||
// this covers any other post-Ready close that looked like success.
|
||||
@@ -76,6 +84,7 @@ pub async fn connect_and_mine(
|
||||
return Err(e);
|
||||
}
|
||||
log::warn!("⛏️ Failed to connect to node: {}", e);
|
||||
metrics::record_connect_failed(); // lair
|
||||
}
|
||||
}
|
||||
|
||||
@@ -145,6 +154,8 @@ async fn handle_connection(
|
||||
let mut cpu_hashes: u64 = 0;
|
||||
let mut gpu_hashes: u64 = 0;
|
||||
let mut result_sent_for_current_job = false;
|
||||
// lair: when the last result went out, for node-attributable idle time.
|
||||
let mut result_sent_at: Option<Instant> = None;
|
||||
|
||||
log::info!("⛏️ Waiting for mining jobs from node...");
|
||||
|
||||
@@ -175,6 +186,13 @@ async fn handle_connection(
|
||||
|
||||
// Check if this result is for the current job (not stale)
|
||||
if worker_result.job_id != internal_job_id {
|
||||
metrics::record_stale_hashes(
|
||||
match worker_result.engine_type {
|
||||
EngineType::Cpu => "CPU",
|
||||
EngineType::Gpu => "GPU",
|
||||
},
|
||||
worker_result.hash_count,
|
||||
); // lair
|
||||
log::debug!(
|
||||
"⏰ Discarding stale result from worker {} (result job_id {} != current {})",
|
||||
worker_result.thread_id,
|
||||
@@ -214,7 +232,15 @@ async fn handle_connection(
|
||||
};
|
||||
|
||||
let msg = MinerMessage::JobResult(result);
|
||||
send_message_checked(&connection, &mut send, &msg).await?;
|
||||
if let Err(e) = send_message_checked(&connection, &mut send, &msg).await {
|
||||
metrics::record_result_send_failed(); // lair
|
||||
return Err(e);
|
||||
}
|
||||
// lair: seal latency and the start of the idle window.
|
||||
metrics::record_result_submitted(
|
||||
worker_result.found_at.map(|t| t.elapsed()),
|
||||
);
|
||||
result_sent_at = Some(Instant::now());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -239,6 +265,10 @@ async fn handle_connection(
|
||||
match msg_result {
|
||||
Ok(MinerMessage::NewJob(request)) => {
|
||||
*authenticated = true;
|
||||
metrics::record_job_received(); // lair
|
||||
if let Some(t) = result_sent_at.take() {
|
||||
metrics::record_job_idle(t.elapsed());
|
||||
}
|
||||
log::info!(
|
||||
"⛏️ Received job: id={}, hash=0x{}",
|
||||
request.job_id,
|
||||
|
||||
Reference in New Issue
Block a user