| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "database/sql" |
| 7 | "database/sql/driver" |
| 8 | "encoding/hex" |
| 9 | "encoding/json" |
| 10 | "errors" |
| 11 | "fmt" |
| 12 | "github.com/carlosframework/platform/internal/hostops" |
| 13 | "log" |
| 14 | "time" |
| 15 | ) |
| 16 | |
| 17 | // These tables exist only in the claim-on console's existing durable DB. |
| 18 | // The lexer blocks all ordinary handles from naming either table. |
| 19 | func initSQLReceipts(db *sql.DB, instance string) error { |
| 20 | _, err := db.Exec(`CREATE TABLE IF NOT EXISTS claim_sql_identity (id TEXT PRIMARY KEY, instance TEXT NOT NULL DEFAULT ''); |
| 21 | CREATE TABLE IF NOT EXISTS claim_sql_receipts (effect TEXT PRIMARY KEY, scope TEXT NOT NULL, state TEXT NOT NULL CHECK(state IN ('committed','fenced')));`) |
| 22 | if err != nil { |
| 23 | return err |
| 24 | } |
| 25 | var hasInstance int |
| 26 | if err = db.QueryRow(`SELECT count(*) FROM pragma_table_info('claim_sql_identity') WHERE name = 'instance'`).Scan(&hasInstance); err != nil { |
| 27 | return err |
| 28 | } |
| 29 | if hasInstance == 0 { |
| 30 | if _, err = db.Exec(`ALTER TABLE claim_sql_identity ADD COLUMN instance TEXT NOT NULL DEFAULT ''`); err != nil { |
| 31 | return err |
| 32 | } |
| 33 | } |
| 34 | var nonce [24]byte |
| 35 | if _, err = rand.Read(nonce[:]); err != nil { |
| 36 | return err |
| 37 | } |
| 38 | if _, err = db.Exec(`INSERT INTO claim_sql_identity(id, instance) SELECT ?, ? WHERE NOT EXISTS(SELECT 1 FROM claim_sql_identity)`, hex.EncodeToString(nonce[:]), instance); err != nil { |
| 39 | return err |
| 40 | } |
| 41 | // A row from before the instance was bound takes this instance. That is |
| 42 | // safe because the instance is part of the database ID: effects bound to |
| 43 | // the old ID stay unsettled here, and a clone adopting the same row gets |
| 44 | // an ID of its own. |
| 45 | if _, err = db.Exec(`UPDATE claim_sql_identity SET instance = ? WHERE instance = ''`, instance); err != nil { |
| 46 | return err |
| 47 | } |
| 48 | var rows int |
| 49 | var stored string |
| 50 | if err = db.QueryRow(`SELECT count(*), COALESCE(MIN(instance), '') FROM claim_sql_identity`).Scan(&rows, &stored); err != nil { |
| 51 | return err |
| 52 | } |
| 53 | if rows != 1 { |
| 54 | return fmt.Errorf("claim SQL receipt database has %d identity rows: %w", rows, hostops.ErrIdentity) |
| 55 | } |
| 56 | if stored == instance { |
| 57 | return nil |
| 58 | } |
| 59 | // Restored onto another instance (review I2-2): adopt it under a fresh |
| 60 | // nonce, so this copy has a database ID no other copy has had, even one |
| 61 | // moved back to an instance it lived on before. Copied receipts still |
| 62 | // cannot settle the original's effects, because every settlement compares |
| 63 | // the effect's database ID first; those effects stay unsettled and are |
| 64 | // alerted past sql_recover_after. |
| 65 | if _, err = rand.Read(nonce[:]); err != nil { |
| 66 | return err |
| 67 | } |
| 68 | if _, err = db.Exec(`UPDATE claim_sql_identity SET id = ?, instance = ?`, hex.EncodeToString(nonce[:]), instance); err != nil { |
| 69 | return err |
| 70 | } |
| 71 | log.Printf("console: claim SQL receipt database moved from instance %s to %s; it has a new database ID, and effects bound to the old one stay unsettled", stored, instance) |
| 72 | return nil |
| 73 | } |
| 74 | |
| 75 | // databaseID is the receipt database's full identity: the stored nonce, the |
| 76 | // instance verified at open, and the physical file, re-read on every call so |
| 77 | // a database replaced under the running process is noticed. |
| 78 | func (g *claimGate) databaseID(nonce string) (string, error) { |
| 79 | physical, err := sqlPhysicalIdentity(g.path) |
| 80 | if err != nil { |
| 81 | return "", err |
| 82 | } |
| 83 | if physical != g.physical || g.instance == "" { |
| 84 | return "", hostops.ErrIdentity |
| 85 | } |
| 86 | return nonce + ":" + g.instance + ":" + physical, nil |
| 87 | } |
| 88 | func cloneSQLReceiptTables(in map[string]claimSQLTable) map[string]claimSQLTable { |
| 89 | out := make(map[string]claimSQLTable, len(in)+2) |
| 90 | for k, v := range in { |
| 91 | out[k] = v |
| 92 | } |
| 93 | out["claim_sql_identity"] = claimSQLTable{NonApp: "private SQL receipt database identity; ordinary SQL handles refuse access"} |
| 94 | out["claim_sql_receipts"] = claimSQLTable{NonApp: "private effect receipts and fences; ordinary SQL handles refuse access"} |
| 95 | return out |
| 96 | } |
| 97 | func (s *Store) sqlDatabaseID() (string, error) { |
| 98 | if s == nil || s.gate == nil || s.rawBoot == nil { |
| 99 | return "", hostops.ErrContract |
| 100 | } |
| 101 | var id string |
| 102 | err := s.rawBoot.QueryRow(`SELECT id FROM claim_sql_identity`).Scan(&id) |
| 103 | if err != nil { |
| 104 | return "", err |
| 105 | } |
| 106 | return s.gate.databaseID(id) |
| 107 | } |
| 108 | |
| 109 | // FenceSQL serializes with the original transaction on SQLite's writer lock. |
| 110 | // Inserting the durable fence even when a receipt is absent closes the |
| 111 | // issued-before-BEGIN crash window. A delayed BEGIN can only refuse it. |
| 112 | func (s *Store) FenceSQL(ctx context.Context, sc hostops.SQLReceiptScope) error { |
| 113 | _, err := s.FenceSQLOutcome(ctx, sc) |
| 114 | return err |
| 115 | } |
| 116 | |
| 117 | // FenceSQLOutcome is FenceSQL that also reports whether the original |
| 118 | // transaction's committed receipt was found, so recovery records what it |
| 119 | // proved rather than calling every outcome a fence. |
| 120 | func (s *Store) FenceSQLOutcome(ctx context.Context, sc hostops.SQLReceiptScope) (committed bool, err error) { |
| 121 | if s.gate == nil || sc.Effect == "" || sc.Operation == "" || sc.AppID == "" || sc.Epoch == 0 { |
| 122 | return false, hostops.ErrIdentity |
| 123 | } |
| 124 | id, err := s.sqlDatabaseID() |
| 125 | if err != nil { |
| 126 | return false, err |
| 127 | } |
| 128 | if id != sc.Database { |
| 129 | return false, hostops.ErrIdentity |
| 130 | } |
| 131 | body, err := json.Marshal(sc) |
| 132 | if err != nil { |
| 133 | return false, err |
| 134 | } |
| 135 | conn, err := s.rawBoot.Conn(ctx) |
| 136 | if err != nil { |
| 137 | return false, err |
| 138 | } |
| 139 | defer conn.Close() |
| 140 | if _, err = conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil { |
| 141 | return false, err |
| 142 | } |
| 143 | defer conn.ExecContext(context.Background(), "ROLLBACK") |
| 144 | var have, state string |
| 145 | err = conn.QueryRowContext(ctx, `SELECT scope, state FROM claim_sql_receipts WHERE effect=?`, sc.Effect).Scan(&have, &state) |
| 146 | if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 147 | return false, err |
| 148 | } |
| 149 | if err == nil && have != string(body) { |
| 150 | return false, hostops.ErrIdentity |
| 151 | } |
| 152 | if _, err = conn.ExecContext(ctx, `INSERT INTO claim_sql_receipts(effect,scope,state) VALUES(?,?,'fenced') ON CONFLICT(effect) DO NOTHING`, sc.Effect, string(body)); err != nil { |
| 153 | return false, err |
| 154 | } |
| 155 | if _, err = conn.ExecContext(ctx, "COMMIT"); err != nil { |
| 156 | return false, err |
| 157 | } |
| 158 | return state == "committed", nil |
| 159 | } |
| 160 | |
| 161 | type claimSavepoint struct{ c *claimConn } |
| 162 | |
| 163 | func (t *claimSavepoint) Commit() error { |
| 164 | return t.c.innerExec(context.Background(), "RELEASE claim_batch_statement") |
| 165 | } |
| 166 | func (t *claimSavepoint) Rollback() error { |
| 167 | if err := t.c.innerExec(context.Background(), "ROLLBACK TO claim_batch_statement"); err != nil { |
| 168 | return err |
| 169 | } |
| 170 | return t.c.innerExec(context.Background(), "RELEASE claim_batch_statement") |
| 171 | } |
| 172 | func (c *claimConn) beginBatch(ctx context.Context, b *claimBinding) error { |
| 173 | // IMMEDIATE is required: a paused deferred read transaction cannot upgrade |
| 174 | // after recovery inserted a fence using another connection. |
| 175 | if err := c.innerExec(ctx, "BEGIN IMMEDIATE"); err != nil { |
| 176 | return err |
| 177 | } |
| 178 | b.outer = &claimBatchTx{c: c} |
| 179 | if len(b.receipts) == 0 { |
| 180 | return nil |
| 181 | } |
| 182 | for _, sc := range b.receipts { |
| 183 | rows, err := c.inner.QueryContext(ctx, `SELECT id FROM claim_sql_identity`, nil) |
| 184 | if err != nil { |
| 185 | return err |
| 186 | } |
| 187 | vals := make([]driver.Value, 1) |
| 188 | err = rows.Next(vals) |
| 189 | rows.Close() |
| 190 | if err != nil { |
| 191 | return err |
| 192 | } |
| 193 | nonce, ok := vals[0].(string) |
| 194 | if !ok { |
| 195 | return hostops.ErrIdentity |
| 196 | } |
| 197 | id, err := c.g.databaseID(nonce) |
| 198 | if err != nil { |
| 199 | return err |
| 200 | } |
| 201 | if id != sc.Database { |
| 202 | return hostops.ErrIdentity |
| 203 | } |
| 204 | // Any prior terminal row makes replay impossible, including a recovery |
| 205 | // fence written before the first SQL call reached this connection. |
| 206 | rows, err = c.inner.QueryContext(ctx, `SELECT count(*) FROM claim_sql_receipts WHERE effect=?`, []driver.NamedValue{{Ordinal: 1, Value: sc.Effect}}) |
| 207 | if err != nil { |
| 208 | return err |
| 209 | } |
| 210 | vals = make([]driver.Value, 1) |
| 211 | err = rows.Next(vals) |
| 212 | rows.Close() |
| 213 | n, _ := vals[0].(int64) |
| 214 | if err != nil { |
| 215 | return err |
| 216 | } |
| 217 | if n != 0 { |
| 218 | return hostops.ErrPending |
| 219 | } |
| 220 | } |
| 221 | return nil |
| 222 | } |
| 223 | |
| 224 | type claimBatchTx struct{ c *claimConn } |
| 225 | |
| 226 | func (t *claimBatchTx) Commit() error { return t.c.innerExec(context.Background(), "COMMIT") } |
| 227 | func (t *claimBatchTx) Rollback() error { return t.c.innerExec(context.Background(), "ROLLBACK") } |
| 228 | func (c *claimConn) endBatch(b *claimBinding, commit bool) error { |
| 229 | if b.outer == nil { |
| 230 | return hostops.ErrContract |
| 231 | } |
| 232 | // The caller sealed every public handle; this runs under the driver lock, |
| 233 | // so nothing else is in flight. Finish what the handles left behind. |
| 234 | b.sealed.Store(true) |
| 235 | c.closeRows() |
| 236 | if c.tx != nil { |
| 237 | c.tx.abort() |
| 238 | if commit { |
| 239 | // A transaction left open means the requested commit cannot be |
| 240 | // honoured: roll back and say so, never report success. |
| 241 | err := fmt.Errorf("%w: transaction still open at batch commit", ErrClaimSQLScope) |
| 242 | if rerr := b.outer.Rollback(); rerr != nil { |
| 243 | c.bad = true |
| 244 | err = errors.Join(err, rerr) |
| 245 | } |
| 246 | b.outer = nil |
| 247 | return err |
| 248 | } |
| 249 | } |
| 250 | if commit { |
| 251 | err := b.lostErr() |
| 252 | if err == nil { |
| 253 | err = b.authorizeCommit(b.ctx, c) |
| 254 | } |
| 255 | if err != nil { |
| 256 | if rerr := b.outer.Rollback(); rerr != nil { |
| 257 | c.bad = true |
| 258 | err = errors.Join(err, rerr) |
| 259 | } |
| 260 | b.outer = nil |
| 261 | return err |
| 262 | } |
| 263 | } |
| 264 | if commit { |
| 265 | for _, receipt := range b.receipts { |
| 266 | body, err := json.Marshal(receipt) |
| 267 | if err != nil { |
| 268 | return err |
| 269 | } |
| 270 | if err = c.innerExec(b.ctx, `INSERT INTO claim_sql_receipts(effect,scope,state) VALUES(?,?,'committed')`, receipt.Effect, string(body)); err != nil { |
| 271 | return err |
| 272 | } |
| 273 | } |
| 274 | } |
| 275 | var err error |
| 276 | if commit { |
| 277 | err = b.outer.Commit() |
| 278 | } else { |
| 279 | err = b.outer.Rollback() |
| 280 | } |
| 281 | b.outer = nil |
| 282 | if err != nil { |
| 283 | c.bad = true |
| 284 | } |
| 285 | return err |
| 286 | } |
| 287 | |
| 288 | // A returned SQL error is not necessarily ambiguous: after handles joined, |
| 289 | // the same positive local fence decides it immediately. Keep the callback's |
| 290 | // error for its caller while settling authority only after this proof. |
| 291 | func (s *Store) finishPreparedSQL(ctx context.Context, p *hostops.PreparedSQLPermit, callErr error) error { |
| 292 | if callErr != nil { |
| 293 | proofCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 20*time.Second) |
| 294 | defer cancel() |
| 295 | if err := s.FenceSQL(proofCtx, p.Receipt()); err != nil { |
| 296 | return errors.Join(err, p.FinishTransaction(callErr)) |
| 297 | } |
| 298 | } |
| 299 | return p.FinishTransaction(nil) |
| 300 | } |
| 301 | |
amadan