| 1 | package registry |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | ) |
| 9 | |
| 10 | // Recovery handovers remain pending until the successor's prelaunch is durable. |
| 11 | // The existing isolated registry transport owns these writes, including at root. |
| 12 | func (r *Registry) BeginRecoveryHandover(ctx context.Context, id, body string) error { |
| 13 | if r.remote != nil { |
| 14 | _, err := r.remote.call(ctx, remoteRequest{Method: "BeginRecoveryHandover", Id: id, Journal: body}, nil) |
| 15 | return err |
| 16 | } |
| 17 | if id == "" || body == "" { |
| 18 | return ErrClaimOwnership |
| 19 | } |
| 20 | if err := r.recoveryJournal(ctx); err != nil { |
| 21 | return err |
| 22 | } |
| 23 | tx, err := r.beginClaimTransaction(ctx, true) |
| 24 | if err != nil { |
| 25 | return err |
| 26 | } |
| 27 | defer tx.Rollback() |
| 28 | var saved string |
| 29 | err = tx.QueryRowContext(ctx, `SELECT body FROM recovery_handovers WHERE id=?`, id).Scan(&saved) |
| 30 | if err == nil { |
| 31 | if saved != body { |
| 32 | return ErrClaimOwnership |
| 33 | } |
| 34 | return nil |
| 35 | } |
| 36 | if !errors.Is(err, sql.ErrNoRows) { |
| 37 | return err |
| 38 | } |
| 39 | if _, err = tx.ExecContext(ctx, `INSERT INTO recovery_handovers(id,body,complete) VALUES(?,?,0)`, id, body); err != nil { |
| 40 | return err |
| 41 | } |
| 42 | return tx.Commit() |
| 43 | } |
| 44 | func (r *Registry) PendingRecoveryHandover(ctx context.Context, id string) (string, error) { |
| 45 | if r.remote != nil { |
| 46 | out, err := r.remote.call(ctx, remoteRequest{Method: "PendingRecoveryHandover", Id: id}, nil) |
| 47 | return out.Text, err |
| 48 | } |
| 49 | if err := r.recoveryJournal(ctx); err != nil { |
| 50 | return "", err |
| 51 | } |
| 52 | var body string |
| 53 | err := r.sql.QueryRowContext(ctx, `SELECT body FROM recovery_handovers WHERE id=? AND complete=0`, id).Scan(&body) |
| 54 | if errors.Is(err, sql.ErrNoRows) { |
| 55 | return "", nil |
| 56 | } |
| 57 | return body, err |
| 58 | } |
| 59 | func (r *Registry) CompleteRecoveryHandover(ctx context.Context, id, body string) error { |
| 60 | if r.remote != nil { |
| 61 | _, err := r.remote.call(ctx, remoteRequest{Method: "CompleteRecoveryHandover", Id: id, Journal: body}, nil) |
| 62 | return err |
| 63 | } |
| 64 | if err := r.recoveryJournal(ctx); err != nil { |
| 65 | return err |
| 66 | } |
| 67 | result, err := r.sql.ExecContext(ctx, `UPDATE recovery_handovers SET complete=1 WHERE id=? AND body=?`, id, body) |
| 68 | if err != nil { |
| 69 | return err |
| 70 | } |
| 71 | n, err := result.RowsAffected() |
| 72 | if err != nil { |
| 73 | return err |
| 74 | } |
| 75 | if n != 1 { |
| 76 | return ErrClaimOwnership |
| 77 | } |
| 78 | return nil |
| 79 | } |
| 80 | func (r *Registry) recoveryJournal(ctx context.Context) error { |
| 81 | _, err := r.sql.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS recovery_handovers(id TEXT PRIMARY KEY,body TEXT NOT NULL,complete INTEGER NOT NULL)`) |
| 82 | return err |
| 83 | } |
| 84 | |
| 85 | func (r *Registry) PendingRecoveryHandovers(ctx context.Context) (map[string]string, error) { |
| 86 | if r.remote != nil { |
| 87 | out, err := r.remote.call(ctx, remoteRequest{Method: "PendingRecoveryHandovers"}, nil) |
| 88 | if err != nil { |
| 89 | return nil, err |
| 90 | } |
| 91 | var bodies map[string]string |
| 92 | err = json.Unmarshal([]byte(out.Text), &bodies) |
| 93 | return bodies, err |
| 94 | } |
| 95 | if err := r.recoveryJournal(ctx); err != nil { |
| 96 | return nil, err |
| 97 | } |
| 98 | rows, err := r.sql.QueryContext(ctx, `SELECT id,body FROM recovery_handovers WHERE complete=0`) |
| 99 | if err != nil { |
| 100 | return nil, err |
| 101 | } |
| 102 | defer rows.Close() |
| 103 | bodies := map[string]string{} |
| 104 | for rows.Next() { |
| 105 | var id, body string |
| 106 | if err := rows.Scan(&id, &body); err != nil { |
| 107 | return nil, err |
| 108 | } |
| 109 | bodies[id] = body |
| 110 | } |
| 111 | return bodies, rows.Err() |
| 112 | } |
| 113 | |
| 114 | func (r *Registry) ReviseRecoveryHandover(ctx context.Context, id, prior, body string) error { |
| 115 | if r.remote != nil { |
| 116 | _, err := r.remote.call(ctx, remoteRequest{Method: "ReviseRecoveryHandover", Id: id, Expected: prior, Journal: body}, nil) |
| 117 | return err |
| 118 | } |
| 119 | if id == "" || prior == "" || body == "" { |
| 120 | return ErrClaimOwnership |
| 121 | } |
| 122 | if err := r.recoveryJournal(ctx); err != nil { |
| 123 | return err |
| 124 | } |
| 125 | tx, err := r.beginClaimTransaction(ctx, true) |
| 126 | if err != nil { |
| 127 | return err |
| 128 | } |
| 129 | defer tx.Rollback() |
| 130 | result, err := tx.ExecContext(ctx, `UPDATE recovery_handovers SET body=? WHERE id=? AND body=? AND complete=0`, body, id, prior) |
| 131 | if err != nil { |
| 132 | return err |
| 133 | } |
| 134 | n, err := result.RowsAffected() |
| 135 | if err != nil { |
| 136 | return err |
| 137 | } |
| 138 | if n != 1 { |
| 139 | return ErrClaimOwnership |
| 140 | } |
| 141 | return tx.Commit() |
| 142 | } |
| 143 | |
amadan