lkb-pipeline/neon_worker.R
allaun e5bfdc0492 feat: LKB compute pipeline + eigensolid receipts
Containerized LKB worker (Dockerfile.lkb), Postgres-backed queue,
auto-deploy timer, and eigensolid convergence check module.

Build: 0 jobs (no Lean build needed)
2026-07-08 20:01:29 -05:00

50 lines
No EOL
2 KiB
R

source("boot.R")
box_use(./core/laurent2)
box_use(./core/lkb)
library(DBI); library(RPostgres)
PSQL <- function(query) {
con <- dbConnect(RPostgres::Postgres(), host=Sys.getenv("PGHOST", "100.79.14.103"),
dbname="research_stack", user="kaleidoscope", password="kaleidoscope")
on.exit(dbDisconnect(con))
dbGetQuery(con, query)
}
cat(sprintf("[%s] neon worker starting (PID %d)\n", Sys.time(), Sys.getpid()), flush = TRUE)
n_jobs <- 0L
repeat {
result <- tryCatch({
PSQL("UPDATE lkb.compute_queue c SET status = 'running', started_at = NOW()
WHERE c.equation_id = (SELECT equation_id FROM lkb.compute_queue
WHERE status = 'pending' ORDER BY word_length ASC NULLS LAST LIMIT 1
FOR UPDATE SKIP LOCKED)
RETURNING c.equation_id, c.n, c.braid_word, c.word_length")
}, error = function(e) data.frame())
if (nrow(result) == 0) break
eq_id <- result$equation_id[1]
n <- result$n[1]
word_str <- result$braid_word[1]
word_len <- if (is.na(result$word_length[1])) 0L else result$word_length[1]
N <- n * (n - 1) / 2L
cat(sprintf(" [%s] B%d word, %d gens (N=%d)\n", eq_id, n, word_len, N), flush = TRUE)
t0 <- Sys.time()
tryCatch({
if (word_str == "ID" || is.na(word_str) || word_str == "") {
entry <- "1"
} else {
steps <- as.integer(strsplit(word_str, ",")[[1]])
gens <- lkb$lkb_generators(n)
mat <- laurent2$l2_mat_identity(N)
for (g in steps) mat <- laurent2$l2_mat_mul(mat, gens[[g]])
entry <- laurent2$l2_str(mat[[1, 3]])
}
esc <- gsub("'", "''", entry)
PSQL(sprintf("UPDATE lkb.compute_queue SET status = 'done', result_entry = '%s',
finished_at = NOW(), steps_done = %d WHERE equation_id = '%s'",
esc, word_len, eq_id))
cat(sprintf(" [%s] done in %.1f s (%d total)\n", eq_id,
difftime(Sys.time(), t0, units="secs"), n_jobs + 1), flush = TRUE)
n_jobs <- n_jobs + 1L
}, error = function(e) {
cat(sprintf(" [%s] ERROR: %s\n", eq_id, conditionMessage(e)), flush = TRUE)
})
}
cat(sprintf("[%s] worker finished -- %d jobs total\n", Sys.time(), n_jobs), flush = TRUE)