open_chess_server/
analyse.rs

1//! The fair-play worker (`open-chess analyse`): takes finished games off the
2//! queue, has Stockfish evaluate every position, and stores a screening
3//! report per human player. With no game to screen, it evaluates positions
4//! people confirmed from board photos for the position dictionary, scoring
5//! every legal move. Run as many as you like (games are claimed with
6//! SKIP LOCKED and a lease, so a crashed worker's game is picked up again).
7
8use std::time::Duration;
9
10use anyhow::{Context, Result};
11use chess_core::Side;
12use chess_engine::{Engine as _, Limit, UciEngine};
13use chess_fairplay::{
14    AnalysisConfig, Model, app_match, app_suggestions, assess_with_app, evaluate_game, features,
15};
16use tokio::sync::watch;
17use tracing::{info, warn};
18
19use crate::config::AnalyseConfig;
20use crate::db::Db;
21use crate::metrics::Metrics;
22use crate::model;
23
24async fn spawn_engine(cfg: &AnalyseConfig) -> Result<UciEngine> {
25    let path = cfg
26        .stockfish
27        .clone()
28        .or_else(chess_engine::find_stockfish)
29        .context("no Stockfish: install it or set CHESS_STOCKFISH")?;
30    let threads = cfg.threads.max(1).to_string();
31    let hash = cfg.hash_mb.max(16).to_string();
32    let e = UciEngine::spawn(&path, &[("Threads", &threads), ("Hash", &hash)]).await?;
33    info!(engine = e.name(), "fair-play engine ready");
34    Ok(e)
35}
36
37/// Claim the next game to screen (or one whose worker died).
38async fn claim(db: &Db) -> Result<Option<String>> {
39    let c = db.conn().await?;
40    Ok(c.query_opt(
41        &db.sql(
42            "UPDATE chess.games SET analysis = 'running', analysis_lease_until = now() + interval '15 minutes'
43              WHERE id = (SELECT id FROM chess.games
44                           WHERE analysis = 'pending' OR (analysis = 'running' AND analysis_lease_until < now())
45                           ORDER BY finished_at LIMIT 1 FOR UPDATE SKIP LOCKED)
46             RETURNING id",
47        ),
48        &[],
49    )
50    .await?
51    .map(|r| r.get(0)))
52}
53
54async fn set_state(db: &Db, id: &str, state: &str) -> Result<()> {
55    let c = db.conn().await?;
56    c.execute(
57        &db.sql("UPDATE chess.games SET analysis = $2, analysis_lease_until = NULL WHERE id = $1"),
58        &[&id, &state],
59    )
60    .await?;
61    Ok(())
62}
63
64async fn screen(
65    db: &Db,
66    metrics: &Metrics,
67    engine: &mut UciEngine,
68    cfg: &AnalyseConfig,
69    screening: &Model,
70    id: &str,
71) -> Result<()> {
72    let c = db.conn().await?;
73    let row = model::game(db, &*c, id).await?.context("game vanished")?;
74    drop(c);
75    let moves: Vec<String> = row.moves.split_whitespace().map(str::to_owned).collect();
76    let acfg = AnalysisConfig { limit: Limit::Depth(cfg.depth), multipv: 3, skip_plies: cfg.skip_plies };
77    let evals = evaluate_game(engine, row.initial_fen.as_deref(), &moves, &row.move_ms, &acfg).await?;
78    // What the training app's engine would have suggested at each move
79    // (CPU-bound, so off the async threads).
80    let app = {
81        let (fen, mv, depth, nodes) = (row.initial_fen.clone(), moves.clone(), cfg.app_depth, cfg.app_nodes);
82        let net = match &cfg.app_net {
83            Some(p) => Some(std::sync::Arc::new(
84                chess_analysis::nnue::Network::from_bytes(
85                    &std::fs::read(p).with_context(|| format!("reading {p}"))?,
86                )
87                .map_err(|e| anyhow::anyhow!("{p}: {e}"))?,
88            )),
89            None => None,
90        };
91        tokio::task::spawn_blocking(move || app_suggestions(fen.as_deref(), &mv, depth, nodes, net)).await??
92    };
93    let c = db.conn().await?;
94    store_positions(db, &*c, row.initial_fen.as_deref(), &evals, engine.name(), cfg.depth).await?;
95    for side in [Side::White, Side::Black] {
96        let player = row.player(side);
97        if player.bot {
98            continue; // bots are engines, openly
99        }
100        let f = features(&evals, side, cfg.skip_plies);
101        let fingerprint = app_match(&evals, &app, side, cfg.skip_plies);
102        let a = assess_with_app(&f, Some(&fingerprint), screening);
103        let declared: bool = c
104            .query_one(
105                &db.sql("SELECT EXISTS (SELECT 1 FROM chess.assist_marks WHERE game_id = $1 AND player_id = $2 AND assist)"),
106                &[&row.id, &player.player_id],
107            )
108            .await?
109            .get(0);
110        let side_evals: Vec<&chess_fairplay::MoveEval> = evals.iter().filter(|e| e.side == side).collect();
111        c.execute(
112            &db.sql(
113                "INSERT INTO chess.fairplay_reports (game_id, side, player_id, score, flagged, moves, accuracy, acpl, top1,
114                        top3, time_cv, time_corr, reasons, engine, depth, evals, app_disagree, app_picked,
115                        app_picked_reference, declared_assist)
116                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)
117                 ON CONFLICT (game_id, side) DO UPDATE SET score = EXCLUDED.score, flagged = EXCLUDED.flagged,
118                        moves = EXCLUDED.moves, accuracy = EXCLUDED.accuracy, acpl = EXCLUDED.acpl, top1 = EXCLUDED.top1,
119                        top3 = EXCLUDED.top3, time_cv = EXCLUDED.time_cv, time_corr = EXCLUDED.time_corr,
120                        reasons = EXCLUDED.reasons, engine = EXCLUDED.engine, depth = EXCLUDED.depth,
121                        evals = EXCLUDED.evals, app_disagree = EXCLUDED.app_disagree, app_picked = EXCLUDED.app_picked,
122                        app_picked_reference = EXCLUDED.app_picked_reference, declared_assist = EXCLUDED.declared_assist,
123                        created_at = now()",
124            ),
125            &[
126                &row.id,
127                &side.name(),
128                &player.player_id,
129                &(a.score as f32),
130                &a.flagged,
131                &(f.moves as i32),
132                &(f.accuracy as f32),
133                &(f.acpl as f32),
134                &(f.top1 as f32),
135                &(f.top3 as f32),
136                &f.time_cv.map(|v| v as f32),
137                &f.time_criticality_corr.map(|v| v as f32),
138                &a.reasons,
139                &engine.name(),
140                &(cfg.depth as i32),
141                &serde_json::to_value(&side_evals)?,
142                &(fingerprint.disagreements as i32),
143                &(fingerprint.picked_app as i32),
144                &(fingerprint.picked_reference as i32),
145                &declared,
146            ],
147        )
148        .await?;
149        metrics.analysed.with_label_values(&[if a.flagged { "yes" } else { "no" }]).inc();
150        if a.flagged {
151            info!(
152                game = row.id,
153                player = player.username,
154                score = a.score,
155                "fair-play flag: {}",
156                a.reasons.join("; ")
157            );
158        }
159    }
160    Ok(())
161}
162
163/// Add every screened position's evaluation to the position dictionary,
164/// keeping the deepest one seen.
165async fn store_positions(
166    db: &Db,
167    c: &impl tokio_postgres::GenericClient,
168    initial_fen: Option<&str>,
169    evals: &[chess_fairplay::MoveEval],
170    engine: &str,
171    depth: u32,
172) -> Result<()> {
173    let mut game = chess_core::Game::new(initial_fen).context("replaying for the position dictionary")?;
174    let sql = db.sql(
175        "INSERT INTO chess.positions (key, best, score_cp, depth, engine) VALUES ($1, $2, $3, $4, $5)
176         ON CONFLICT (key) DO UPDATE SET seen = chess.positions.seen + 1,
177             best = CASE WHEN EXCLUDED.depth >= chess.positions.depth THEN EXCLUDED.best ELSE chess.positions.best END,
178             score_cp = CASE WHEN EXCLUDED.depth >= chess.positions.depth THEN EXCLUDED.score_cp ELSE chess.positions.score_cp END,
179             engine = CASE WHEN EXCLUDED.depth >= chess.positions.depth THEN EXCLUDED.engine ELSE chess.positions.engine END,
180             depth = GREATEST(EXCLUDED.depth, chess.positions.depth), updated_at = now()",
181    );
182    for e in evals {
183        let key = crate::http::lab::position_key(&game);
184        c.execute(&sql, &[&key, &e.best, &e.best_cp, &(depth as i32), &engine]).await?;
185        if game.play_uci(&e.uci).is_err() {
186            break;
187        }
188    }
189    Ok(())
190}
191
192/// The most positions' moves a dictionary evaluation scores (every legal
193/// move, up to Stockfish's MultiPV limit).
194const MAX_MULTIPV: u32 = 256;
195/// Mates as centipawns, on the scale the screened positions are stored
196/// with (chess-fairplay's).
197const MATE_CP: i32 = 1000;
198
199/// Evaluate one position waiting in the dictionary (from a photo): the
200/// best move and score, and every legal move's score. `false` when there
201/// was nothing to do.
202async fn evaluate_queued(db: &Db, engine: &mut UciEngine, cfg: &AnalyseConfig) -> Result<bool> {
203    let c = db.conn().await?;
204    let Some(key) = c
205        .query_opt(
206            &db.sql(
207                "UPDATE chess.positions SET eval_lease_until = now() + interval '10 minutes'
208                  WHERE key = (SELECT key FROM chess.positions
209                                WHERE score_cp IS NULL AND (eval_lease_until IS NULL OR eval_lease_until < now())
210                                ORDER BY updated_at LIMIT 1 FOR UPDATE SKIP LOCKED)
211                 RETURNING key",
212            ),
213            &[],
214        )
215        .await?
216        .map(|r| r.get::<_, String>(0))
217    else {
218        return Ok(false);
219    };
220    drop(c);
221    let fen = chess_analysis::dictionary::fen_of_key(&key);
222    let legal = chess_core::Game::new(Some(&fen)).map_or(0, |g| g.legal_moves().len());
223    if legal == 0 {
224        // Not a position anyone can play on: drop it.
225        let c = db.conn().await?;
226        c.execute(&db.sql("DELETE FROM chess.positions WHERE key = $1 AND score_cp IS NULL"), &[&key])
227            .await?;
228        return Ok(true);
229    }
230    let multipv = u32::try_from(legal).unwrap_or(MAX_MULTIPV).min(MAX_MULTIPV);
231    engine.new_game().await?;
232    let a = engine.analyse(&fen, &[], Limit::Depth(cfg.depth), multipv).await?;
233    let Some(top) = a.lines.first() else { anyhow::bail!("no lines for {key}") };
234    let moves: Vec<serde_json::Value> = a
235        .lines
236        .iter()
237        .filter_map(|l| {
238            Some(serde_json::json!({"uci": l.pv.first()?, "cp": l.score.to_cp(MATE_CP), "depth": l.depth}))
239        })
240        .collect();
241    let best = a.best_move.clone().or_else(|| top.pv.first().cloned());
242    let c = db.conn().await?;
243    c.execute(
244        &db.sql(
245            "UPDATE chess.positions SET best = $2, score_cp = $3, depth = $4, engine = $5, moves = $6,
246                    eval_lease_until = NULL, updated_at = now()
247              WHERE key = $1",
248        ),
249        &[
250            &key,
251            &best,
252            &top.score.to_cp(MATE_CP),
253            &(top.depth as i32),
254            &engine.name(),
255            &serde_json::Value::from(moves),
256        ],
257    )
258    .await?;
259    info!(position = key, moves = legal, "dictionary position evaluated");
260    Ok(true)
261}
262
263pub async fn run(
264    db: Db,
265    metrics: Metrics,
266    cfg: AnalyseConfig,
267    mut stop: watch::Receiver<bool>,
268) -> Result<()> {
269    let (screening, configured) = cfg.model()?;
270    if !configured {
271        warn!(
272            "no CHESS_FAIRPLAY_MODEL: screening with the untuned example model; flags are only illustrative"
273        );
274    }
275    let mut engine = spawn_engine(&cfg).await?;
276    loop {
277        if *stop.borrow() {
278            break;
279        }
280        let id = match claim(&db).await {
281            Ok(id) => id,
282            Err(e) => {
283                warn!("fair-play queue: {e:#}");
284                None
285            }
286        };
287        let Some(id) = id else {
288            // Nothing to screen: a photo position for the dictionary.
289            match evaluate_queued(&db, &mut engine, &cfg).await {
290                Ok(true) => continue,
291                Ok(false) => {}
292                Err(e) => {
293                    warn!("dictionary position: {e:#}");
294                    engine = spawn_engine(&cfg).await?;
295                }
296            }
297            if cfg.once {
298                break;
299            }
300            tokio::select! {
301                () = tokio::time::sleep(Duration::from_secs(5)) => {}
302                () = crate::stopped(&mut stop) => break,
303            }
304            continue;
305        };
306        let started = std::time::Instant::now();
307        match screen(&db, &metrics, &mut engine, &cfg, &screening, &id).await {
308            Ok(()) => {
309                set_state(&db, &id, "done").await?;
310                info!(game = id, secs = started.elapsed().as_secs(), "game screened");
311            }
312            Err(e) => {
313                warn!(game = id, "screening failed: {e:#}");
314                set_state(&db, &id, "failed").await?;
315                // The engine may be the problem: start a fresh one.
316                engine = spawn_engine(&cfg).await?;
317            }
318        }
319        if cfg.once {
320            break;
321        }
322    }
323    engine.quit().await;
324    Ok(())
325}
open_chess_server/analyse.rs

open_chess_server/
analyse.rs
