| 1 | package registry |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "reflect" |
| 10 | ) |
| 11 | |
| 12 | // RebindClaimRows transfers an exact stopped installation's registry rows. |
| 13 | // The caller holds hostguard and proves the source cut and successor admission. |
| 14 | // A transaction preserves row contents and refuses any concurrent replacement. |
| 15 | func (r *Registry) RebindClaimRows(ctx context.Context, old, next ColdRows) error { |
| 16 | if r.remote != nil { |
| 17 | _, err := r.remote.call(ctx, remoteRequest{Method: "RebindClaimRows", Old: old, NextValue: next}, nil) |
| 18 | return err |
| 19 | } |
| 20 | if _, err := r.sql.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS recovery_rows(id TEXT PRIMARY KEY,body BLOB NOT NULL)`); 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 | id := fmt.Sprintf("%s/%s/%d", next.Route.Host, next.Route.ClaimIncarnation, next.Route.ClaimStamp) |
| 29 | var saved []byte |
| 30 | err = tx.QueryRowContext(ctx, `SELECT body FROM recovery_rows WHERE id=?`, id).Scan(&saved) |
| 31 | recorded := err == nil |
| 32 | if recorded { |
| 33 | var rows struct{ Old, Next ColdRows } |
| 34 | if json.Unmarshal(saved, &rows) != nil || withoutBuildStateCopy(rows.Old.Route) != withoutBuildStateCopy(old.Route) || !reflect.DeepEqual(coldRowsWithoutBuildStateCopy(rows.Next), coldRowsWithoutBuildStateCopy(next)) { |
| 35 | return ErrClaimOwnership |
| 36 | } |
| 37 | old = rows.Old |
| 38 | } else if !errors.Is(err, sql.ErrNoRows) { |
| 39 | return err |
| 40 | } |
| 41 | if !old.Route.ClaimOwnership.Valid() || !next.Route.ClaimOwnership.Valid() || old.Route.Host != next.Route.Host || old.Route.ClaimIncarnation != next.Route.ClaimIncarnation || next.Route.ClaimStamp <= old.Route.ClaimStamp || len(old.Timers) != len(next.Timers) { |
| 42 | return ErrClaimOwnership |
| 43 | } |
| 44 | a, b := withoutBuildStateCopy(old.Route), withoutBuildStateCopy(next.Route) |
| 45 | a.ClaimOwnership, b.ClaimOwnership = ClaimOwnership{}, ClaimOwnership{} |
| 46 | if a != b { |
| 47 | return ErrClaimOwnership |
| 48 | } |
| 49 | for i, t := range next.Timers { |
| 50 | prior := old.Timers[i] |
| 51 | if !prior.ClaimOwnership.Valid() || prior.Host != old.Route.Host || prior.ClaimIncarnation != old.Route.ClaimIncarnation || prior.ClaimStamp != old.Route.ClaimStamp || prior.ClaimOwner != old.Route.ClaimOwner { |
| 52 | return ErrClaimOwnership |
| 53 | } |
| 54 | a, b := prior, t |
| 55 | a.ClaimOwnership, b.ClaimOwnership = ClaimOwnership{}, ClaimOwnership{} |
| 56 | if a != b || !t.ClaimOwnership.Valid() || t.Host != next.Route.Host || t.ClaimIncarnation != next.Route.ClaimIncarnation || t.ClaimStamp != next.Route.ClaimStamp || t.ClaimOwner != next.Route.ClaimOwner { |
| 57 | return ErrClaimOwnership |
| 58 | } |
| 59 | } |
| 60 | row, err := scanRoute(tx.QueryRowContext(ctx, `SELECT `+routeCols+` FROM routes WHERE host=?`, old.Route.Host).Scan) |
| 61 | if err != nil { |
| 62 | return err |
| 63 | } |
| 64 | row = withoutBuildStateCopy(row) |
| 65 | replay := row == withoutBuildStateCopy(next.Route) |
| 66 | if !replay && row != withoutBuildStateCopy(old.Route) { |
| 67 | return ErrClaimOwnership |
| 68 | } |
| 69 | rows, err := tx.conn.QueryContext(ctx, `SELECT `+timerCols+` FROM timers WHERE host=? ORDER BY name`, old.Route.Host) |
| 70 | if err != nil { |
| 71 | return err |
| 72 | } |
| 73 | var timers []Timer |
| 74 | for rows.Next() { |
| 75 | t, e := scanTimer(rows) |
| 76 | if e != nil { |
| 77 | rows.Close() |
| 78 | return e |
| 79 | } |
| 80 | timers = append(timers, t) |
| 81 | } |
| 82 | err = rows.Err() |
| 83 | rows.Close() |
| 84 | if err != nil { |
| 85 | return err |
| 86 | } |
| 87 | want := old.Timers |
| 88 | if replay { |
| 89 | want = next.Timers |
| 90 | } |
| 91 | if len(timers) != len(want) { |
| 92 | return ErrClaimOwnership |
| 93 | } |
| 94 | for _, t := range timers { |
| 95 | found := false |
| 96 | for _, w := range want { |
| 97 | if reflect.DeepEqual(t, w) { |
| 98 | found = true |
| 99 | break |
| 100 | } |
| 101 | } |
| 102 | if !found { |
| 103 | return ErrClaimOwnership |
| 104 | } |
| 105 | } |
| 106 | if replay { |
| 107 | if !recorded { |
| 108 | return ErrClaimOwnership |
| 109 | } |
| 110 | return nil |
| 111 | } |
| 112 | o := next.Route.ClaimOwnership |
| 113 | if _, err = tx.ExecContext(ctx, `UPDATE routes SET claim_incarnation=?,claim_stamp=?,claim_owner=?,claim_row_id=? WHERE host=?`, o.ClaimIncarnation, o.ClaimStamp, o.ClaimOwner, o.ClaimRowID, next.Route.Host); err != nil { |
| 114 | return err |
| 115 | } |
| 116 | for _, t := range next.Timers { |
| 117 | o := t.ClaimOwnership |
| 118 | if _, err = tx.ExecContext(ctx, `UPDATE timers SET claim_incarnation=?,claim_stamp=?,claim_owner=?,claim_row_id=? WHERE host=? AND name=?`, o.ClaimIncarnation, o.ClaimStamp, o.ClaimOwner, o.ClaimRowID, t.Host, t.Name); err != nil { |
| 119 | return err |
| 120 | } |
| 121 | } |
| 122 | body, err := json.Marshal(struct{ Old, Next ColdRows }{old, next}) |
| 123 | if err != nil { |
| 124 | return err |
| 125 | } |
| 126 | if _, err = tx.ExecContext(ctx, `INSERT INTO recovery_rows(id,body) VALUES(?,?)`, id, body); err != nil { |
| 127 | return err |
| 128 | } |
| 129 | return tx.Commit() |
| 130 | } |
| 131 | |
amadan