WIP lot V : le tampon câblé dans le serveur (migration 0010, simulation, discard-simulation), tests en cours de reprise — repris tel quel après la coupure du 02/11

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Yvv
2026-11-03 05:22:24 +01:00
co-authored by Claude Opus 5.5
parent ee37c3b4d5
commit 4b81a2e909
16 changed files with 1503 additions and 213 deletions
+534 -123
View File
@@ -21,6 +21,21 @@
//! Ğecko code). A ballot is read one last time at its end (24 hours), then let
//! go; the adoption pass runs after each confirmation. Ballots let go are purged
//! after a while (a day for those the wallet never signed, a week for the others).
//!
//! **The buffer** (ADR 0012, Yvv 2026-11-02: « je vois ce que j'ai fait sans
//! attendre », with a control « a posteriori » of the integrity of the vote): a
//! pending ballot is an **intention**, seen at once by the person (« ton vote »)
//! and in the totals shown, never counted for adoption; the chain brings its
//! **proof**; the **reconciliation** (`ld_engine::buffer::reconcile`) keeps one
//! voice per identity and per version, the last in the order of the chain, and
//! sets aside what is never proven in time, refused, overtaken — or a *for*
//! proven while an objection holds its version back. What is set aside is told
//! in the box « Messages ».
//!
//! **The simulation**: while the chain is closed (no `LD_VOTE_ACCOUNT`), a group
//! of the chain votes anyway, without a wallet — through the same buffer: a
//! sealed ballot, simulated, proven at once by the simulated confirmer, for the
//! identity of the session.
use crate::AppState;
use crate::adoption::{self, Entered};
@@ -28,6 +43,7 @@ use crate::auth::{IdentityRow, User};
use crate::docs::{self, DocState};
use crate::error::{AppError, AppResult};
use crate::indexer::{self, Found, Reading};
use crate::messages::{self, Topic};
use crate::util;
use axum::Json;
use axum::extract::{Path, State};
@@ -35,10 +51,13 @@ use axum::http::{HeaderValue, header};
use axum::response::{IntoResponse, Response};
use ld_engine::address;
use ld_engine::ballot::{self, ChainPosition, Choice, Expected, Verdict};
use ld_engine::buffer::{self, Outcome, Proven, Stage};
use rand::RngCore;
use serde::Deserialize;
use serde_json::json;
use sqlx::SqliteConnection;
use std::collections::{HashMap, HashSet};
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio::sync::Mutex;
use trustin_access::VoteWay;
@@ -167,7 +186,62 @@ pub(crate) async fn write_server_votes(state: &AppState, identity: &str, casts:
Ok(())
}
/// `POST /api/docs/{id}/votes` — a grouped vote in a managed group: one write.
/// How a vote of this session is written by the server in a group now.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum ServerWay {
/// A managed group: the server is the authority, the vote is recorded at once.
Direct,
/// A group of the chain while the chain is closed: a simulated ballot, proven
/// by the simulated confirmer (ADR 0012).
Simulated,
}
/// Where the vote goes through the chain and the chain is open, never on the
/// server (`409 chain_only`): a ballot to sign.
pub(crate) fn server_way(state: &AppState, group: &docs::GroupRow) -> AppResult<ServerWay> {
match trustin_access::vote_way(&group.to_group()) {
VoteWay::Server => Ok(ServerWay::Direct),
VoteWay::Chain if state.chain().is_some() => Err(AppError::Conflict("chain_only")),
VoteWay::Chain => Ok(ServerWay::Simulated),
}
}
/// May this session write its vote on the server, this way? trustinWallet's rules:
/// `votes_in` (a managed group), `casts_simulated` (the simulation).
pub(crate) async fn may_write(
state: &AppState,
user: &IdentityRow,
group: &docs::GroupRow,
way: ServerWay,
) -> AppResult<bool> {
let granted = docs::granted(state, &group.slug, &user.id).await?;
let (id, g) = (user.to_identity(), group.to_group());
Ok(match way {
ServerWay::Direct => trustin_access::votes_in(&id, &g, granted),
ServerWay::Simulated => trustin_access::casts_simulated(&id, &g, granted, state.chain().is_some()),
})
}
/// Writes the votes of a session on the server — directly, or in simulation — and
/// runs the adoption pass. Returns what entered into force.
pub(crate) async fn write(
state: &AppState,
user: &IdentityRow,
st: &DocState,
way: ServerWay,
casts: &[Cast],
) -> AppResult<Vec<Entered>> {
match way {
ServerWay::Direct => {
write_server_votes(state, &user.id, casts).await?;
adoption::run(state, &st.doc.id).await
}
ServerWay::Simulated => simulate(state, &user.id, st, casts).await,
}
}
/// `POST /api/docs/{id}/votes` — a grouped vote written by the server: one write
/// in a managed group, one simulated ballot while the chain is closed.
pub async fn vote_many(
State(state): State<AppState>,
User(user): User,
@@ -176,16 +250,84 @@ pub async fn vote_many(
) -> AppResult<Json<serde_json::Value>> {
let st = docs::load_state(&state, &doc_id).await?;
docs::require_open(&state, &user, &st.group).await?;
if trustin_access::vote_way(&st.group.to_group()) == VoteWay::Chain {
return Err(AppError::Conflict("chain_only"));
}
if !docs::may_vote(&state, &user, &st.group).await? {
let way = server_way(&state, &st.group)?;
if !may_write(&state, &user, &st.group, way).await? {
return Err(AppError::Forbidden);
}
let casts = casts(&st, &b.lines)?;
write_server_votes(&state, &user.id, &casts).await?;
let entered = adoption::run(&state, &st.doc.id).await?;
Ok(Json(json!({ "count": casts.len(), "entered": entered })))
let entered = write(&state, &user, &st, way, &casts).await?;
Ok(Json(json!({ "count": casts.len(), "entered": entered, "simulated": way == ServerWay::Simulated })))
}
/// The simulated confirmer (ADR 0012): while the chain is closed, a vote goes
/// through the same buffer as a vote of the chain — a sealed ballot, simulated
/// (no account, no amount, no wallet), proven at once for the identity of the
/// session, at block 0, then reconciled. Returns what entered into force.
pub(crate) async fn simulate(
state: &AppState,
requester: &str,
st: &DocState,
casts: &[Cast],
) -> AppResult<Vec<Entered>> {
let lines: Vec<ballot::Line> = casts
.iter()
.map(|c| ballot::Line {
proposal_id: c.proposal_id.clone(),
revision: u32::try_from(c.revision).unwrap_or(u32::MAX),
choice: c.choice,
})
.collect();
let salt = salt();
let sealed = ballot::seal(&st.doc.id, None, &lines, &salt).map_err(|_| AppError::Invalid("lines"))?;
let id = util::id();
let mut tx = state.db.begin().await?;
sqlx::query(
"INSERT INTO ballots (id, doc_id, requested_by, author, salt, digest, remark_hash, account, amount, status,
created_at, expires_at, mode)
VALUES (?, ?, ?, NULL, ?, ?, ?, '', 0, 'pending', ?, ?, 'simulated')",
)
.bind(&id)
.bind(&st.doc.id)
.bind(requester)
.bind(&salt)
.bind(&sealed.digest)
.bind(&sealed.remark_hash)
.bind(util::now())
.bind(util::in_seconds(EXPIRES_HOURS * 3600))
.execute(&mut *tx)
.await?;
for c in casts {
sqlx::query("INSERT INTO ballot_lines (ballot_id, proposal_id, revision, choice) VALUES (?, ?, ?, ?)")
.bind(&id)
.bind(&c.proposal_id)
.bind(c.revision)
.bind(c.choice.as_str())
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
let row = load(state, &id).await?;
settle_simulated(state, &row).await
}
/// Proves a simulated ballot: the voice of the session that asked for it.
pub async fn settle_simulated(state: &AppState, row: &BallotRow) -> AppResult<Vec<Entered>> {
settle(state, row, Proof { voter: row.requested_by.clone(), position: None, found: None }).await
}
#[derive(Deserialize)]
pub struct DevChainBody {
open: bool,
}
/// `POST /api/dev/vote-chain` — tests only, with the dev login: closes (or opens
/// again) the chain of this server, to try the simulation (ADR 0012).
pub async fn dev_chain(State(state): State<AppState>, Json(b): Json<DevChainBody>) -> AppResult<Json<serde_json::Value>> {
if !state.cfg.dev_login {
return Err(AppError::NotFound);
}
state.chain_closed.store(!b.open, Ordering::Relaxed);
Ok(Json(json!({ "open": state.chain().is_some() })))
}
// ---------------------------------------------------------------------------
@@ -210,15 +352,34 @@ pub struct BallotRow {
pub created_at: String,
pub expires_at: String,
pub checked_at: Option<String>,
/// `chain`, or `simulated` (ADR 0012); read as `chain` when absent.
#[sqlx(default)]
pub mode: String,
#[sqlx(default)]
pub settled_at: Option<String>,
}
const BALLOT_SELECT: &str = "SELECT id, doc_id, requested_by, author, digest, remark_hash, account, amount, status,
reason, voter_id, block, entered, created_at, expires_at, checked_at FROM ballots";
reason, voter_id, block, entered, created_at, expires_at, checked_at, mode, settled_at FROM ballots";
impl BallotRow {
pub fn comment(&self) -> String {
format!("{}{}", ballot::PREFIX, self.digest)
}
/// A ballot of the simulation, proven by the simulated confirmer.
pub fn simulated(&self) -> bool {
self.mode == "simulated"
}
pub fn mode(&self) -> &'static str {
if self.simulated() { "simulated" } else { "chain" }
}
/// Where it sits on the chain: never for a simulated ballot (block 0 orders it only).
pub fn block_shown(&self) -> Option<i64> {
self.block.filter(|_| !self.simulated())
}
}
fn salt() -> String {
@@ -244,7 +405,7 @@ pub async fn create(
if !trustin_access::asks_ballot(&user.to_identity(), &group, granted) {
return Err(AppError::Forbidden);
}
let chain = state.cfg.vote.as_ref().ok_or(AppError::Conflict("chain_vote_off"))?;
let chain = state.chain().ok_or(AppError::Conflict("chain_vote_off"))?;
let casts = casts(&st, &b.lines)?;
let author = match b.author.as_deref().map(str::trim).filter(|a| !a.is_empty()) {
Some(a) => Some(address::to_g1(a).ok_or(AppError::Invalid("author"))?),
@@ -252,7 +413,7 @@ pub async fn create(
};
// Pending, or let go within the hour: letting a ballot go does not free its place at once.
let held: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ballots WHERE requested_by = ?
"SELECT COUNT(*) FROM ballots WHERE requested_by = ? AND mode = 'chain'
AND (status = 'pending' OR (reason = 'dropped' AND settled_at > ?))",
)
.bind(&user.id)
@@ -329,7 +490,9 @@ pub async fn status(
if row.requested_by != user.id {
return Err(AppError::NotFound);
}
if row.status == "pending" && row.checked_at.as_ref().is_none_or(|c| *c < util::in_seconds(-RECHECK_SECS)) {
if row.status == "pending"
&& (row.simulated() || row.checked_at.as_ref().is_none_or(|c| *c < util::in_seconds(-RECHECK_SECS)))
{
check(&state, &row).await?;
row = load(&state, &id).await?;
}
@@ -386,9 +549,10 @@ pub async fn mine(State(state): State<AppState>, User(user): User) -> AppResult<
lines: i64,
created_at: String,
settled_at: Option<String>,
mode: String,
}
let rows = sqlx::query_as::<_, Row>(&format!(
"SELECT b.id, b.doc_id, d.group_slug, d.slug, d.title, b.digest, b.status, b.reason, b.block,
"SELECT b.id, b.doc_id, d.group_slug, d.slug, d.title, b.digest, b.status, b.reason, b.block, b.mode,
i.display_name AS voter, (SELECT COUNT(*) FROM ballot_lines l WHERE l.ballot_id = b.id) AS lines,
b.created_at, b.settled_at
FROM ballots b JOIN documents d ON d.id = b.doc_id LEFT JOIN identities i ON i.id = b.voter_id
@@ -404,7 +568,9 @@ pub async fn mine(State(state): State<AppState>, User(user): User) -> AppResult<
json!({
"id": r.id, "doc": { "id": r.doc_id, "group": r.group_slug, "slug": r.slug, "title": title_json(&r.title) },
"comment": format!("{}{}", ballot::PREFIX, r.digest), "status": r.status, "reason": r.reason,
"block": r.block, "voter": r.voter, "lines": r.lines, "createdAt": r.created_at, "settledAt": r.settled_at,
"block": r.block.filter(|_| r.mode != "simulated"), "voter": r.voter, "lines": r.lines,
"createdAt": r.created_at, "settledAt": r.settled_at,
"mode": if r.mode == "simulated" { "simulated" } else { "chain" },
})
})
.collect();
@@ -495,7 +661,8 @@ pub async fn receipt(State(state): State<AppState>, User(user): User, Path(id):
"amount": row.amount,
"status": row.status,
"reason": row.reason,
"block": row.block,
"mode": row.mode(),
"block": row.block_shown(),
"event": extra.event_id,
"signer": extra.signer,
"voter": voter,
@@ -527,14 +694,17 @@ async fn view(state: &AppState, row: &BallotRow) -> AppResult<serde_json::Value>
row.entered.as_deref().and_then(|e| serde_json::from_str(e).ok()).unwrap_or_else(|| json!([]));
Ok(json!({
"id": row.id, "status": row.status, "reason": row.reason, "comment": row.comment(),
"account": row.account, "amount": row.amount, "author": row.author, "block": row.block,
"voter": voter, "entered": entered, "expiresAt": row.expires_at,
"account": row.account, "amount": row.amount, "author": row.author, "block": row.block_shown(),
"voter": voter, "entered": entered, "expiresAt": row.expires_at, "mode": row.mode(),
}))
}
/// Reads the indexer for one pending ballot and settles it if the chain has spoken
/// — in a final block. At its end, a ballot is read one last time, then let go.
pub async fn check(state: &AppState, row: &BallotRow) -> AppResult<()> {
if row.simulated() {
return settle_simulated(state, row).await.map(|_| ());
}
let now = util::now();
let due = row.expires_at <= now;
let past_grace = row.expires_at <= util::in_seconds(-GRACE_SECS);
@@ -576,7 +746,9 @@ pub async fn check(state: &AppState, row: &BallotRow) -> AppResult<()> {
match verdict {
// The right to vote is checked now, at confirmation, by trustinWallet's rule.
Verdict::Paid { identity } if trustin_access::chain_vote_counts(identity.member) => {
confirm(state, row, found, identity.index).await
let proof =
Proof { voter: format!("idty:{}", identity.index), position: Some(found.comment.position), found: Some(found) };
settle(state, row, proof).await.map(|_| ())
}
Verdict::Paid { .. } => refuse(state, row, found, "not_member").await,
other => refuse(state, row, found, other.reason().unwrap_or("refused")).await,
@@ -599,153 +771,256 @@ async fn refuse(state: &AppState, row: &BallotRow, found: &Found, reason: &str)
.await?;
if done.rows_affected() == 1 {
tracing::info!(ballot = %row.id, reason, "ballot refused");
tell(state, &row.id, &row.requested_by, &row.doc_id, "ballot_refused").await?;
}
Ok(())
}
/// The ballot counts: written for the identity of the signer, line by line,
/// where no more recent ballot of that identity decided already. A ballot sealed
/// before an objection counts as it was signed — a *for* too, as a vote already
/// cast —, and the version held back still cannot enter (ADR 0010).
async fn confirm(state: &AppState, row: &BallotRow, found: &Found, index: u32) -> AppResult<()> {
let voter = format!("idty:{index}");
let position = found.comment.position;
/// Tells the session that asked for a ballot what became of it (ADR 0012): a
/// message of its box, without gesture — one votes again in the text.
async fn tell(state: &AppState, ballot_id: &str, requester: &str, doc_id: &str, kind: &str) -> AppResult<()> {
messages::notify(state, requester, doc_id, Topic::Ballot, ballot_id, kind, 1, None).await
}
/// The proof of a ballot: whose voice it is, and where it sits on the chain.
pub struct Proof<'a> {
/// `idty:<index>` of the signer; for a simulated ballot, the identity of the
/// session that asked (there is no wallet).
pub voter: String,
/// `None`: a simulated ballot — block 0, at the next rank of the simulation.
pub position: Option<ChainPosition>,
/// What the chain said of the comment (account, name, event); none in simulation.
pub found: Option<&'a Found>,
}
/// The ballot is proven: confirmed for its voter, then each of its versions is
/// reconciled (`reconcile`). A *for* proven while an objection holds its
/// version back is set aside, and told (ADR 0012, revising 0010: it no longer
/// counts « as it was signed »). Proving twice changes nothing. Returns what
/// entered into force.
async fn settle(state: &AppState, row: &BallotRow, proof: Proof<'_>) -> AppResult<Vec<Entered>> {
let now = util::now();
{
// The versions an objection holds back now (read before the transaction: one writer).
let held = docs::load_state(state, &row.doc_id).await?.blocked;
let mut set_aside = false;
let position = {
let _one = CONFIRM.lock().await;
let mut tx = state.db.begin().await?;
// The identity of the signer, as the chain has it now: a member (just
// checked), with its name and its account. Undone with the rest if the
// ballot was settled in the meantime.
let admin = state.cfg.admin_g1_ids.contains(&voter);
sqlx::query(
"INSERT INTO identities (id, kind, display_name, address, g1_member, admin, created_at)
VALUES (?, 'g1', ?, ?, 1, ?, ?)
ON CONFLICT(id) DO UPDATE SET display_name = excluded.display_name, address = excluded.address,
g1_member = 1, admin = MAX(identities.admin, excluded.admin)",
)
.bind(&voter)
.bind(found.name.clone().unwrap_or_else(|| voter.clone()))
.bind(&found.signer)
.bind(admin)
.bind(&now)
.execute(&mut *tx)
.await?;
if let Some(found) = proof.found {
// The identity of the signer, as the chain has it now: a member (just
// checked), with its name and its account. Undone with the rest if the
// ballot was settled in the meantime.
let admin = state.cfg.admin_g1_ids.contains(&proof.voter);
sqlx::query(
"INSERT INTO identities (id, kind, display_name, address, g1_member, admin, created_at)
VALUES (?, 'g1', ?, ?, 1, ?, ?)
ON CONFLICT(id) DO UPDATE SET display_name = excluded.display_name, address = excluded.address,
g1_member = 1, admin = MAX(identities.admin, excluded.admin)",
)
.bind(&proof.voter)
.bind(found.name.clone().unwrap_or_else(|| proof.voter.clone()))
.bind(&found.signer)
.bind(admin)
.bind(&now)
.execute(&mut *tx)
.await?;
}
let position = match proof.position {
Some(p) => p,
None => {
let last: i64 = sqlx::query_scalar(
"SELECT IFNULL(MAX(event_index), 0) FROM ballots WHERE mode = 'simulated' AND status = 'confirmed'",
)
.fetch_one(&mut *tx)
.await?;
ChainPosition { block: 0, event: u32::try_from(last.saturating_add(1)).unwrap_or(u32::MAX) }
}
};
let won = sqlx::query(
"UPDATE ballots SET status = 'confirmed', voter_id = ?, signer = ?, block = ?, event_index = ?,
event_id = ?, settled_at = ? WHERE id = ? AND status = 'pending'",
)
.bind(&voter)
.bind(&found.signer)
.bind(&proof.voter)
.bind(proof.found.map(|f| &f.signer))
.bind(i64::from(position.block))
.bind(i64::from(position.event))
.bind(&found.event_id)
.bind(proof.found.map(|f| &f.event_id))
.bind(&now)
.bind(&row.id)
.execute(&mut *tx)
.await?;
if won.rows_affected() != 1 {
// Settled in the meantime: confirming twice changes nothing.
// Settled in the meantime: proving twice changes nothing.
tx.rollback().await?;
return Ok(());
return Ok(Vec::new());
}
let lines: Vec<(String, i64, String)> =
sqlx::query_as("SELECT proposal_id, revision, choice FROM ballot_lines WHERE ballot_id = ?")
let lines: Vec<(String, String)> =
sqlx::query_as("SELECT proposal_id, choice FROM ballot_lines WHERE ballot_id = ?")
.bind(&row.id)
.fetch_all(&mut *tx)
.await?;
for (proposal_id, revision, choice) in lines {
let others: Vec<(i64, i64)> = sqlx::query_as(
"SELECT b.block, b.event_index FROM ballots b JOIN ballot_lines l ON l.ballot_id = b.id
WHERE b.status = 'confirmed' AND b.voter_id = ? AND l.proposal_id = ? AND b.id <> ?
AND b.block IS NOT NULL AND b.event_index IS NOT NULL",
)
.bind(&voter)
.bind(&proposal_id)
.bind(&row.id)
.fetch_all(&mut *tx)
.await?;
let others: Vec<ChainPosition> = others
.into_iter()
.filter_map(|(b, e)| {
Some(ChainPosition { block: u32::try_from(b).ok()?, event: u32::try_from(e).ok()? })
})
.collect();
if !ballot::decides(position, &others) {
for (proposal_id, choice) in &lines {
let Some(choice) = Choice::parse(choice) else {
tracing::warn!(ballot = %row.id, %choice, "unknown choice in a ballot line");
continue;
}
match Choice::parse(&choice) {
Some(Choice::None) => {
sqlx::query("DELETE FROM votes WHERE proposal_id = ? AND identity_id = ?")
.bind(&proposal_id)
.bind(&voter)
.execute(&mut *tx)
.await?;
}
Some(c) => {
sqlx::query(
"INSERT INTO votes (proposal_id, identity_id, choice, revision, cast_at, source, block, event_index, ballot_id)
VALUES (?, ?, ?, ?, ?, 'chain', ?, ?, ?)
ON CONFLICT(proposal_id, identity_id) DO UPDATE SET choice = excluded.choice,
revision = excluded.revision, cast_at = excluded.cast_at, source = 'chain',
block = excluded.block, event_index = excluded.event_index, ballot_id = excluded.ballot_id",
)
.bind(&proposal_id)
.bind(&voter)
.bind(c.as_str())
.bind(revision)
.bind(&now)
.bind(i64::from(position.block))
.bind(i64::from(position.event))
};
if buffer::discarded_by_objection(choice, held.contains(proposal_id)) {
sqlx::query("UPDATE ballot_lines SET outcome = ? WHERE ballot_id = ? AND proposal_id = ?")
.bind(Outcome::Objection.as_str())
.bind(&row.id)
.bind(proposal_id)
.execute(&mut *tx)
.await?;
}
None => tracing::warn!(ballot = %row.id, %choice, "unknown choice in a ballot line"),
set_aside = true;
}
}
for (proposal_id, _) in &lines {
reconcile(&mut tx, &proof.voter, proposal_id, &now).await?;
}
tx.commit().await?;
}
tracing::info!(ballot = %row.id, block = position.block, event = position.event, "ballot confirmed");
position
};
tracing::info!(ballot = %row.id, mode = row.mode(), block = position.block, event = position.event, "ballot proven");
let entered: Vec<Entered> = adoption::run(state, &row.doc_id).await?;
sqlx::query("UPDATE ballots SET entered = ? WHERE id = ?")
.bind(serde_json::to_string(&entered).unwrap_or_else(|_| "[]".into()))
.bind(&row.id)
.execute(&state.db)
.await?;
if set_aside {
tell(state, &row.id, &row.requested_by, &row.doc_id, "ballot_objection").await?;
}
Ok(entered)
}
/// The reconciliation of one identity on one version (ADR 0012): every proven
/// line of it, read again; one voice, the last in the order of the chain
/// (`ld_engine::buffer::reconcile`), whatever the order the proofs came in. Each
/// line is told what became of it; the voice that stands is written in `votes`
/// — or, every line set aside, the vote of before is left as it is.
pub async fn reconcile(conn: &mut SqliteConnection, voter: &str, proposal_id: &str, now: &str) -> AppResult<()> {
#[derive(sqlx::FromRow)]
struct Row {
ballot_id: String,
mode: String,
block: Option<i64>,
event_index: Option<i64>,
revision: i64,
choice: String,
outcome: Option<String>,
}
let rows = sqlx::query_as::<_, Row>(
"SELECT b.id AS ballot_id, b.mode, b.block, b.event_index, l.revision, l.choice, l.outcome
FROM ballots b JOIN ballot_lines l ON l.ballot_id = b.id
WHERE b.status = 'confirmed' AND b.voter_id = ? AND l.proposal_id = ?
ORDER BY b.block, b.event_index, b.created_at, b.rowid",
)
.bind(voter)
.bind(proposal_id)
.fetch_all(&mut *conn)
.await?;
let mut kept: Vec<(&Row, Proven)> = Vec::with_capacity(rows.len());
for r in &rows {
let position = r.block.zip(r.event_index).and_then(|(b, e)| {
Some(ChainPosition { block: u32::try_from(b).ok()?, event: u32::try_from(e).ok()? })
});
let (Some(position), Some(choice)) = (position, Choice::parse(&r.choice)) else {
tracing::warn!(ballot = %r.ballot_id, "a proven line that does not read back");
continue;
};
let held = r.outcome.as_deref() == Some(Outcome::Objection.as_str());
kept.push((r, Proven { position, choice, held }));
}
let lines: Vec<Proven> = kept.iter().map(|(_, p)| *p).collect();
let done = buffer::reconcile(&lines);
for ((r, _), outcome) in kept.iter().zip(&done.outcomes) {
if r.outcome.as_deref() != Some(outcome.as_str()) {
sqlx::query("UPDATE ballot_lines SET outcome = ? WHERE ballot_id = ? AND proposal_id = ?")
.bind(outcome.as_str())
.bind(&r.ballot_id)
.bind(proposal_id)
.execute(&mut *conn)
.await?;
}
}
let Some((r, line)) = done.standing.and_then(|i| kept.get(i)) else { return Ok(()) };
match line.choice {
Choice::None => {
sqlx::query("DELETE FROM votes WHERE proposal_id = ? AND identity_id = ?")
.bind(proposal_id)
.bind(voter)
.execute(&mut *conn)
.await?;
}
c => {
// Only when it changes: the same ballot standing again keeps its row as it is.
sqlx::query(
"INSERT INTO votes (proposal_id, identity_id, choice, revision, cast_at, source, block, event_index, ballot_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(proposal_id, identity_id) DO UPDATE SET choice = excluded.choice,
revision = excluded.revision, cast_at = excluded.cast_at, source = excluded.source,
block = excluded.block, event_index = excluded.event_index, ballot_id = excluded.ballot_id
WHERE votes.ballot_id IS NOT excluded.ballot_id",
)
.bind(proposal_id)
.bind(voter)
.bind(c.as_str())
.bind(r.revision)
.bind(now)
.bind(if r.mode == "simulated" { "simulated" } else { "chain" })
.bind(i64::from(line.position.block))
.bind(i64::from(line.position.event))
.bind(&r.ballot_id)
.execute(&mut *conn)
.await?;
}
}
Ok(())
}
/// Lets go of one pending ballot not seen on the chain in time.
async fn let_go(state: &AppState, row: &BallotRow) -> AppResult<()> {
sqlx::query("UPDATE ballots SET status = 'expired', settled_at = ? WHERE id = ? AND status = 'pending'")
let done = sqlx::query("UPDATE ballots SET status = 'expired', settled_at = ? WHERE id = ? AND status = 'pending'")
.bind(util::now())
.bind(&row.id)
.execute(&state.db)
.await?;
if done.rows_affected() == 1 {
tell(state, &row.id, &row.requested_by, &row.doc_id, "ballot_expired").await?;
}
Ok(())
}
/// Lets go of the pending ballots past their end and the grace after it: those
/// the indexer could not be read for in time.
pub async fn expire(state: &AppState) -> AppResult<u64> {
let done = sqlx::query(
"UPDATE ballots SET status = 'expired', settled_at = ? WHERE status = 'pending' AND expires_at <= ?",
let gone: Vec<(String, String, String)> = sqlx::query_as(
"UPDATE ballots SET status = 'expired', settled_at = ?
WHERE status = 'pending' AND mode = 'chain' AND expires_at <= ?
RETURNING id, requested_by, doc_id",
)
.bind(util::now())
.bind(util::in_seconds(-GRACE_SECS))
.execute(&state.db)
.fetch_all(&state.db)
.await?;
Ok(done.rows_affected())
for (id, requester, doc_id) in &gone {
tell(state, id, requester, doc_id, "ballot_expired").await?;
}
Ok(u64::try_from(gone.len()).unwrap_or(u64::MAX))
}
/// Purges the ballots let go long enough ago, lines and all. A ballot let go
/// never counted: nothing of a vote is lost.
/// Purges the ballots let go long enough ago, lines and all, and the messages
/// that told of them. A ballot let go never counted: nothing of a vote is lost.
pub async fn purge(state: &AppState) -> AppResult<u64> {
const WHICH: &str = "SELECT id FROM ballots WHERE status = 'expired' AND settled_at IS NOT NULL
AND ((reason = 'dropped' AND settled_at < ?) OR settled_at < ?)";
let (dropped, expired) = (util::in_seconds(-PURGE_DROPPED_SECS), util::in_seconds(-PURGE_EXPIRED_SECS));
let mut tx = state.db.begin().await?;
sqlx::query(&format!("DELETE FROM messages WHERE topic = 'ballot' AND subject_id IN ({WHICH})"))
.bind(&dropped)
.bind(&expired)
.execute(&mut *tx)
.await?;
sqlx::query(&format!("DELETE FROM ballot_lines WHERE ballot_id IN ({WHICH})"))
.bind(&dropped)
.bind(&expired)
@@ -766,8 +1041,19 @@ pub async fn purge(state: &AppState) -> AppResult<u64> {
pub async fn sweep(state: &AppState) -> AppResult<usize> {
expire(state).await?;
purge(state).await?;
// The simulated confirmer: a simulated ballot is proven in its own request; one
// left waiting (a stop in between) is proven here.
let simulated =
sqlx::query_as::<_, BallotRow>(&format!("{BALLOT_SELECT} WHERE status = 'pending' AND mode = 'simulated'"))
.fetch_all(&state.db)
.await?;
for row in &simulated {
if let Err(e) = settle_simulated(state, row).await {
tracing::warn!(ballot = %row.id, error = %e, "simulated ballot");
}
}
let rows = sqlx::query_as::<_, BallotRow>(&format!(
"{BALLOT_SELECT} WHERE status = 'pending'
"{BALLOT_SELECT} WHERE status = 'pending' AND mode = 'chain'
ORDER BY checked_at IS NOT NULL, checked_at, created_at, rowid LIMIT {PER_PASS}"
))
.fetch_all(&state.db)
@@ -823,8 +1109,9 @@ pub async fn my_votes(state: &AppState, me: &str, doc_id: &str) -> AppResult<Has
Ok(rows.into_iter().map(|v| (v.proposal_id.clone(), v)).collect())
}
/// What a ballot of the person, not yet counted, says of one version: pending on
/// the chain, or refused, or let go.
/// What a ballot of the person, not counted, says of one version: pending on the
/// chain (an intention, ADR 0012), refused, let go — or its *for* set aside by an
/// objection (`refused`, reason `objection`).
#[derive(Clone, Debug, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MyBallot {
@@ -832,12 +1119,16 @@ pub struct MyBallot {
pub status: String,
pub reason: Option<String>,
pub choice: String,
/// The revision the line votes.
#[serde(skip)]
pub revision: i64,
}
/// Per version: the most recent pending ballot of `me` asked after its last
/// confirmed one; else its most recent ballot if that one was refused or let go
/// (a confirmed one shows as its vote). A ballot the wallet never signed
/// (dropped) says nothing; one asked before a confirmed vote says nothing either.
/// confirmed one; else its most recent ballot if that one was refused or let go,
/// or proven with its line set aside by an objection (a confirmed one shows as its
/// vote). A ballot the wallet never signed (dropped) says nothing; one asked
/// before a confirmed vote says nothing either.
pub async fn my_ballots(state: &AppState, me: &str, doc_id: &str) -> AppResult<HashMap<String, MyBallot>> {
#[derive(sqlx::FromRow)]
struct Row {
@@ -846,9 +1137,11 @@ pub async fn my_ballots(state: &AppState, me: &str, doc_id: &str) -> AppResult<H
status: String,
reason: Option<String>,
choice: String,
revision: i64,
outcome: Option<String>,
}
let rows = sqlx::query_as::<_, Row>(
"SELECT l.proposal_id, b.id, b.status, b.reason, l.choice FROM ballots b
"SELECT l.proposal_id, b.id, b.status, b.reason, l.choice, l.revision, l.outcome FROM ballots b
JOIN ballot_lines l ON l.ballot_id = b.id
WHERE b.doc_id = ? AND b.requested_by = ? AND IFNULL(b.reason, '') <> 'dropped'
ORDER BY b.created_at, b.rowid",
@@ -860,7 +1153,14 @@ pub async fn my_ballots(state: &AppState, me: &str, doc_id: &str) -> AppResult<H
let mut latest: HashMap<String, MyBallot> = HashMap::new();
let mut pending: HashMap<String, MyBallot> = HashMap::new();
for r in rows {
let b = MyBallot { id: r.id, status: r.status, reason: r.reason, choice: r.choice };
let set_aside = r.status == "confirmed" && r.outcome.as_deref() == Some(Outcome::Objection.as_str());
let b = MyBallot {
id: r.id,
status: if set_aside { "refused".into() } else { r.status },
reason: if set_aside { Some(Outcome::Objection.as_str().into()) } else { r.reason },
choice: r.choice,
revision: r.revision,
};
match b.status.as_str() {
// A confirmed vote: what was asked before it says nothing more.
"confirmed" => {
@@ -871,6 +1171,9 @@ pub async fn my_ballots(state: &AppState, me: &str, doc_id: &str) -> AppResult<H
pending.insert(r.proposal_id, b);
}
_ => {
if set_aside {
pending.remove(&r.proposal_id);
}
latest.insert(r.proposal_id, b);
}
}
@@ -879,29 +1182,137 @@ pub async fn my_ballots(state: &AppState, me: &str, doc_id: &str) -> AppResult<H
Ok(latest)
}
/// The latest intention of a session on a version, still waiting for its proof.
pub struct Intent {
pub requester: String,
pub proposal_id: String,
pub revision: i64,
pub choice: Choice,
}
/// The intentions of a document (ADR 0012): for each session and each version,
/// its most recent pending ballot of the chain asked after its last confirmed one
/// on that version — the rule of `my_ballots`, for everyone.
pub async fn intents(state: &AppState, doc_id: &str) -> AppResult<Vec<Intent>> {
#[derive(sqlx::FromRow)]
struct Row {
requested_by: String,
proposal_id: String,
revision: i64,
choice: String,
}
let rows = sqlx::query_as::<_, Row>(
"SELECT b.requested_by, l.proposal_id, l.revision, l.choice FROM ballots b
JOIN ballot_lines l ON l.ballot_id = b.id
WHERE b.doc_id = ? AND b.status = 'pending' AND b.mode = 'chain'
AND NOT EXISTS (SELECT 1 FROM ballots c JOIN ballot_lines cl ON cl.ballot_id = c.id
WHERE c.requested_by = b.requested_by AND cl.proposal_id = l.proposal_id
AND c.status = 'confirmed'
AND (c.created_at > b.created_at OR (c.created_at = b.created_at AND c.rowid > b.rowid)))
ORDER BY b.created_at, b.rowid",
)
.bind(doc_id)
.fetch_all(&state.db)
.await?;
let mut latest: HashMap<(String, String), Intent> = HashMap::new();
for r in rows {
let Some(choice) = Choice::parse(&r.choice) else { continue };
let key = (r.requested_by.clone(), r.proposal_id.clone());
latest.insert(key, Intent { requester: r.requested_by, proposal_id: r.proposal_id, revision: r.revision, choice });
}
Ok(latest.into_values().collect())
}
/// « Ton vote » as the person sees it now (ADR 0012): its latest gesture — an
/// intention still waiting for the chain — or the vote that counts for it.
pub struct Seen {
pub choice: Option<String>,
/// `recorded`, `pending`, `simulated`, or none.
pub state: Option<&'static str>,
/// It votes a revision gone by: « to reconfirm ».
pub stale: bool,
/// Proven on the chain.
pub chain: bool,
}
pub fn seen(revision: i64, vote: Option<&MyVote>, ballot: Option<&MyBallot>) -> Seen {
if let Some(b) = ballot.filter(|b| b.status == "pending") {
let choice = (b.choice != Choice::None.as_str()).then(|| b.choice.clone());
return Seen {
stale: choice.is_some() && b.revision != revision,
choice,
state: Some(Stage::Pending.as_str()),
chain: false,
};
}
match vote {
Some(v) => Seen {
choice: Some(v.choice.clone()),
state: Some(Stage::of_source(&v.source).as_str()),
stale: v.revision != revision,
chain: v.source == "chain",
},
None => Seen { choice: None, state: None, stale: false, chain: false },
}
}
/// A ballot, for a message of the box of the session that asked for it (ADR
/// 0012): what became of it, line by line. No gesture: one votes again in the text.
pub async fn for_message(state: &AppState, viewer: &IdentityRow, id: &str) -> AppResult<Option<serde_json::Value>> {
#[derive(sqlx::FromRow)]
struct LineRow {
proposal_id: String,
unit_id: String,
code: String,
title: String,
choice: String,
outcome: Option<String>,
}
let row = sqlx::query_as::<_, BallotRow>(&format!("{BALLOT_SELECT} WHERE id = ?"))
.bind(id)
.fetch_optional(&state.db)
.await?;
let Some(row) = row.filter(|r| r.requested_by == viewer.id) else { return Ok(None) };
let lines = sqlx::query_as::<_, LineRow>(
"SELECT l.proposal_id, p.unit_id, u.code, p.title, l.choice, l.outcome FROM ballot_lines l
JOIN proposals p ON p.id = l.proposal_id JOIN units u ON u.id = p.unit_id
WHERE l.ballot_id = ? ORDER BY u.position, u.id",
)
.bind(id)
.fetch_all(&state.db)
.await?;
Ok(Some(json!({
"id": row.id, "status": row.status, "reason": row.reason, "mode": row.mode(),
"createdAt": row.created_at, "settledAt": row.settled_at,
"lines": lines.iter().map(|l| json!({
"proposalId": l.proposal_id, "unitId": l.unit_id, "code": l.code, "title": l.title,
"choice": l.choice, "outcome": l.outcome,
})).collect::<Vec<_>>(),
})))
}
/// How this session votes in a group, and whether it may now.
pub async fn access(state: &AppState, user: &IdentityRow, group: &docs::GroupRow) -> AppResult<serde_json::Value> {
let g = group.to_group();
let granted = docs::granted(state, &g.slug, &user.id).await?;
let id = user.to_identity();
let way = trustin_access::vote_way(&g);
let (open, reason) = match way {
VoteWay::Server => {
let open = trustin_access::votes_in(&id, &g, granted);
(open, (!open).then_some("members_only"))
}
VoteWay::Chain => match (&state.cfg.vote, trustin_access::asks_ballot(&id, &g, granted)) {
(None, _) => (false, Some("chain_off")),
(Some(_), false) => (false, Some("members_only")),
(Some(_), true) => (true, None),
},
// While the chain is closed, a group of the chain votes in simulation: the vote
// leaves the page by a write of the server, marked as simulated (ADR 0012).
let simulated = way == VoteWay::Chain && state.chain().is_none();
let open = match way {
VoteWay::Server => trustin_access::votes_in(&id, &g, granted),
VoteWay::Chain if simulated => trustin_access::casts_simulated(&id, &g, granted, false),
VoteWay::Chain => trustin_access::asks_ballot(&id, &g, granted),
};
let chain = match (way, &state.cfg.vote) {
let reason = (!open).then_some("members_only");
let chain = match (way, state.chain()) {
(VoteWay::Chain, Some(c)) => json!({
"account": c.account, "amount": c.amount, "nodes": c.nodes,
"test": c.test_node.is_some(), "testNode": c.test_node, "expiresHours": EXPIRES_HOURS,
}),
_ => serde_json::Value::Null,
};
Ok(json!({ "way": way, "open": open, "reason": reason, "chain": chain }))
let leaves = if simulated { VoteWay::Server } else { way };
Ok(json!({ "way": leaves, "open": open, "simulated": simulated, "reason": reason, "chain": chain }))
}