| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "database/sql/driver" |
| 7 | "errors" |
| 8 | "github.com/carlosframework/platform/internal/hostops" |
| 9 | "runtime" |
| 10 | "strings" |
| 11 | "sync/atomic" |
| 12 | "testing" |
| 13 | "time" |
| 14 | ) |
| 15 | |
| 16 | // plan9SealRetainedStore is a retained *Store from a batch callback writing |
| 17 | // after the callback returned. |
| 18 | func plan9SealRetainedStore(v *Store, out chan<- error) { |
| 19 | _, err := v.db.Exec(`UPDATE audit SET action='late-store' WHERE account_id=7 AND app='blog'`) |
| 20 | out <- err |
| 21 | } |
| 22 | |
| 23 | // plan9SealQueuedDriver is an operation that already passed the view's public |
| 24 | // check and is queued on the held driver connection. |
| 25 | func plan9SealQueuedDriver(d *claimBoundDB, out chan<- error) { |
| 26 | _, err := d.conn.ExecContext(context.Background(), `UPDATE audit SET action='late-driver' WHERE account_id=7 AND app='blog'`) |
| 27 | out <- err |
| 28 | } |
| 29 | |
| 30 | // plan9SealQueuedStmtExec and plan9SealQueuedStmtQuery run an escaped prepared |
| 31 | // statement: one that was never registered with the view, as if the statement |
| 32 | // had been prepared before the callback returned. |
| 33 | func plan9SealQueuedStmtExec(st *sql.Stmt, out chan<- error) { |
| 34 | _, err := st.ExecContext(context.Background()) |
| 35 | out <- err |
| 36 | } |
| 37 | |
| 38 | func plan9SealQueuedStmtQuery(st *sql.Stmt, out chan<- error) { |
| 39 | rows, err := st.QueryContext(context.Background()) |
| 40 | if err == nil { |
| 41 | for rows.Next() { |
| 42 | } |
| 43 | err = rows.Err() |
| 44 | rows.Close() |
| 45 | } |
| 46 | out <- err |
| 47 | } |
| 48 | |
| 49 | // waitPlan9SealQueued returns once the goroutine running fn has finished |
| 50 | // (done has a value) or is parked on the database/sql driver-connection lock, |
| 51 | // which the batch commit holds. |
| 52 | func waitPlan9SealQueued(t *testing.T, fn string, done chan error) { |
| 53 | t.Helper() |
| 54 | deadline := time.Now().Add(10 * time.Second) |
| 55 | buf := make([]byte, 1<<20) |
| 56 | for time.Now().Before(deadline) { |
| 57 | if len(done) > 0 { |
| 58 | return |
| 59 | } |
| 60 | n := runtime.Stack(buf, true) |
| 61 | for _, g := range strings.Split(string(buf[:n]), "\n\n") { |
| 62 | onDriver := strings.Contains(g, "database/sql.withLock(") || strings.Contains(g, "database/sql.resultFromStatement(") || strings.Contains(g, "database/sql.rowsiFromStatement(") |
| 63 | if strings.Contains(g, "console."+fn+"(") && onDriver && strings.Contains(g, "sync.(*Mutex).lockSlow(") { |
| 64 | return |
| 65 | } |
| 66 | } |
| 67 | runtime.Gosched() |
| 68 | } |
| 69 | t.Fatalf("%s neither finished nor queued on the driver connection", fn) |
| 70 | } |
| 71 | |
| 72 | // TestPlan9SQLBatchSealsRetainedWriteQueuedAcrossCommit checks the view fast |
| 73 | // path and queues driver and prepared-statement writes across COMMIT. None |
| 74 | // may run under the transaction permit once its commit begins. |
| 75 | func TestPlan9SQLBatchSealsRetainedWriteQueuedAcrossCommit(t *testing.T) { |
| 76 | s := openGatedStore(t) |
| 77 | seedGateApp(t, s) |
| 78 | r := receiptFixture(t, s) |
| 79 | var view *Store |
| 80 | var returned atomic.Bool |
| 81 | store, driver := make(chan error, 1), make(chan error, 1) |
| 82 | stmtExec, stmtQuery := make(chan error, 1), make(chan error, 1) |
| 83 | var escapedExec, escapedQuery *sql.Stmt |
| 84 | local := func(ctx context.Context, _ claimSQLScope) error { |
| 85 | if !returned.CompareAndSwap(true, false) { |
| 86 | return ctx.Err() |
| 87 | } |
| 88 | // The commit authorization: endBatch holds the driver connection. |
| 89 | go plan9SealRetainedStore(view, store) |
| 90 | waitPlan9SealQueued(t, "plan9SealRetainedStore", store) |
| 91 | go plan9SealQueuedDriver(view.db.(*claimBoundDB), driver) |
| 92 | waitPlan9SealQueued(t, "plan9SealQueuedDriver", driver) |
| 93 | go plan9SealQueuedStmtExec(escapedExec, stmtExec) |
| 94 | waitPlan9SealQueued(t, "plan9SealQueuedStmtExec", stmtExec) |
| 95 | go plan9SealQueuedStmtQuery(escapedQuery, stmtQuery) |
| 96 | waitPlan9SealQueued(t, "plan9SealQueuedStmtQuery", stmtQuery) |
| 97 | return nil |
| 98 | } |
| 99 | err := s.withPreparedClaimSQL(gateCtx(t), []claimSQLScope{gateScope}, allowAllClaimSQL, local, []hostops.SQLReceiptScope{r}, true, func(v *Store) error { |
| 100 | view = v |
| 101 | // Prepared on the raw sql.Conn, bypassing the view's registration, so |
| 102 | // these statements are past it and only the driver gate can stop them. |
| 103 | d := v.db.(*claimBoundDB) |
| 104 | var perr error |
| 105 | if escapedExec, perr = d.conn.PrepareContext(context.Background(), `UPDATE audit SET action='late-stmt-exec' WHERE account_id=7 AND app='blog'`); perr != nil { |
| 106 | return perr |
| 107 | } |
| 108 | if escapedQuery, perr = d.conn.PrepareContext(context.Background(), `UPDATE audit SET action='late-stmt-query' WHERE account_id=7 AND app='blog' RETURNING id`); perr != nil { |
| 109 | return perr |
| 110 | } |
| 111 | _, err := v.db.Exec(`UPDATE audit SET action='batch' WHERE account_id=7 AND app='blog'`) |
| 112 | returned.Store(true) |
| 113 | return err |
| 114 | }) |
| 115 | if err != nil { |
| 116 | t.Fatal(err) |
| 117 | } |
| 118 | if escapedExec != nil { |
| 119 | defer escapedExec.Close() |
| 120 | } |
| 121 | if escapedQuery != nil { |
| 122 | defer escapedQuery.Close() |
| 123 | } |
| 124 | if returned.Load() { |
| 125 | t.Fatal("commit authorization never ran the local check") |
| 126 | } |
| 127 | for name, ch := range map[string]chan error{ |
| 128 | "retained store (view fast path)": store, |
| 129 | "queued driver": driver, |
| 130 | "escaped stmt Exec": stmtExec, |
| 131 | "escaped stmt QueryContext": stmtQuery, |
| 132 | } { |
| 133 | select { |
| 134 | case werr := <-ch: |
| 135 | if !errors.Is(werr, ErrClaimSQLRevoked) { |
| 136 | t.Errorf("%s write across COMMIT: err=%v, want ErrClaimSQLRevoked", name, werr) |
| 137 | } |
| 138 | case <-time.After(10 * time.Second): |
| 139 | t.Fatalf("%s write never finished", name) |
| 140 | } |
| 141 | } |
| 142 | if n := countWhere(t, s, "audit", "action LIKE 'late-%'"); n != 0 { |
| 143 | t.Errorf("%d retained writes landed outside the receipted transaction", n) |
| 144 | } |
| 145 | if countWhere(t, s, "audit", "action='batch'") != 1 { |
| 146 | t.Error("batch write did not commit") |
| 147 | } |
| 148 | var state string |
| 149 | if err := s.rawBoot.QueryRow(`SELECT state FROM claim_sql_receipts WHERE effect=?`, r.Effect).Scan(&state); err != nil || state != "committed" { |
| 150 | t.Fatalf("receipt=%q err=%v", state, err) |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | // TestPlan9SQLEndBatchLeftoverTxRefusesCommit drives the driver directly: a |
| 155 | // transaction that registered after the handles were joined forces the |
| 156 | // requested commit to roll back, and endBatch must say so instead of |
| 157 | // reporting success with nothing committed. |
| 158 | func TestPlan9SQLEndBatchLeftoverTxRefusesCommit(t *testing.T) { |
| 159 | s := openGatedStore(t) |
| 160 | seedGateApp(t, s) |
| 161 | r := receiptFixture(t, s) |
| 162 | ctx := gateCtx(t) |
| 163 | conn, err := s.rawPool.Conn(ctx) |
| 164 | if err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | defer conn.Close() |
| 168 | sqids := make([]string, 1) |
| 169 | if s.gate.needSqid() { |
| 170 | if err := conn.QueryRowContext(ctx, `SELECT sqid FROM accounts WHERE id = ?`, gateScope.AccountID).Scan(&sqids[0]); err != nil { |
| 171 | t.Fatal(err) |
| 172 | } |
| 173 | } |
| 174 | scopes := []claimSQLScope{gateScope} |
| 175 | b := &claimBinding{ctx: ctx, scopes: scopes, check: func(ctx context.Context, _ claimSQLScope) error { return ctx.Err() }, deadline: time.Now().Add(hostops.SQLTransactionLifetime), batch: true, receipts: []hostops.SQLReceiptScope{r}, dead: make([]bool, 1)} |
| 176 | var endErr error |
| 177 | err = conn.Raw(func(dc any) error { |
| 178 | c := dc.(*claimConn) |
| 179 | if err := c.bindScopes(ctx, b, sqids); err != nil { |
| 180 | return err |
| 181 | } |
| 182 | defer func() { _ = c.unbind(b) }() |
| 183 | if err := c.beginBatch(ctx, b); err != nil { |
| 184 | return err |
| 185 | } |
| 186 | if _, err := c.ExecContext(ctx, `UPDATE audit SET action='batch' WHERE account_id=7 AND app='blog'`, nil); err != nil { |
| 187 | return err |
| 188 | } |
| 189 | if _, err := c.BeginTx(ctx, driver.TxOptions{}); err != nil { |
| 190 | return err |
| 191 | } |
| 192 | endErr = c.endBatch(b, true) |
| 193 | return nil |
| 194 | }) |
| 195 | if err != nil { |
| 196 | t.Fatal(err) |
| 197 | } |
| 198 | if err := conn.Close(); err != nil { |
| 199 | t.Fatal(err) |
| 200 | } |
| 201 | if endErr == nil { |
| 202 | t.Fatal("endBatch reported success although a leftover transaction forced the commit to roll back") |
| 203 | } |
| 204 | if !errors.Is(endErr, ErrClaimSQLScope) { |
| 205 | t.Errorf("endBatch err=%v, want ErrClaimSQLScope", endErr) |
| 206 | } |
| 207 | var n int |
| 208 | if err := s.rawBoot.QueryRow(`SELECT count(*) FROM claim_sql_receipts WHERE effect=?`, r.Effect).Scan(&n); err != nil || n != 0 { |
| 209 | t.Errorf("receipt rows=%d err=%v, want none", n, err) |
| 210 | } |
| 211 | if got := countWhere(t, s, "audit", "action='batch'"); got != 0 { |
| 212 | t.Errorf("batch write present (%d rows) after forced rollback", got) |
| 213 | } |
| 214 | } |
| 215 | |
| 216 | // Seal while a handle has left the driver but has not entered the view's |
| 217 | // registry. It must be closed by the registering goroutine, not escape the |
| 218 | // closeHandles snapshot. Holding mu makes this boundary deterministic. |
| 219 | func TestPlan9SQLSealDuringHandleRegistration(t *testing.T) { |
| 220 | for _, method := range []string{"PrepareContext", "BeginTx"} { |
| 221 | t.Run(method, func(t *testing.T) { |
| 222 | s := openGatedStore(t) |
| 223 | seedGateApp(t, s) |
| 224 | r := receiptFixture(t, s) |
| 225 | err := s.withPreparedClaimSQL(gateCtx(t), []claimSQLScope{gateScope}, allowAllClaimSQL, allowAllClaimSQL, []hostops.SQLReceiptScope{r}, true, func(v *Store) error { |
| 226 | d := v.db.(*claimBoundDB) |
| 227 | d.mu.Lock() |
| 228 | locked := true |
| 229 | defer func() { |
| 230 | if locked { |
| 231 | d.mu.Unlock() |
| 232 | } |
| 233 | }() |
| 234 | done := make(chan error, 1) |
| 235 | go func() { |
| 236 | if method == "PrepareContext" { |
| 237 | st, err := d.PrepareContext(context.Background(), "SELECT 1") |
| 238 | if st != nil { |
| 239 | _ = st.Close() |
| 240 | } |
| 241 | done <- err |
| 242 | } else { |
| 243 | tx, err := d.BeginTx(context.Background(), nil) |
| 244 | if tx != nil { |
| 245 | _ = tx.Rollback() |
| 246 | } |
| 247 | done <- err |
| 248 | } |
| 249 | }() |
| 250 | deadline := time.Now().Add(10 * time.Second) |
| 251 | buf := make([]byte, 1<<20) |
| 252 | waiting := false |
| 253 | for time.Now().Before(deadline) && !waiting { |
| 254 | n := runtime.Stack(buf, true) |
| 255 | for _, g := range strings.Split(string(buf[:n]), "\n\n") { |
| 256 | if strings.Contains(g, "console.(*claimBoundDB)."+method+"(") && strings.Contains(g, "sync.(*Mutex).lockSlow(") { |
| 257 | waiting = true |
| 258 | break |
| 259 | } |
| 260 | } |
| 261 | runtime.Gosched() |
| 262 | } |
| 263 | if !waiting { |
| 264 | t.Fatal("handle never reached registration") |
| 265 | } |
| 266 | d.b.sealed.Store(true) |
| 267 | d.mu.Unlock() |
| 268 | locked = false |
| 269 | select { |
| 270 | case err := <-done: |
| 271 | if !errors.Is(err, ErrClaimSQLRevoked) { |
| 272 | t.Errorf("late registration = %v, want revoked", err) |
| 273 | } |
| 274 | case <-time.After(10 * time.Second): |
| 275 | t.Fatal("late registration did not join") |
| 276 | } |
| 277 | return nil |
| 278 | }) |
| 279 | if err != nil { |
| 280 | t.Fatal(err) |
| 281 | } |
| 282 | }) |
| 283 | } |
| 284 | } |
| 285 | |
amadan