| 1 | package registry |
| 2 | |
| 3 | import ( |
| 4 | "crypto/rand" |
| 5 | "database/sql" |
| 6 | "encoding/hex" |
| 7 | "errors" |
| 8 | ) |
| 9 | |
| 10 | var ErrTimerChanged = errors.New("registry: timer registration changed") |
| 11 | |
| 12 | // UpsertTimerRegistration compares the captured registration and returns the |
| 13 | // resulting registration in the same transaction. Empty means absent at read. |
| 14 | func (r *Registry) UpsertTimerRegistration(t Timer, expected string) (*Timer, error) { |
| 15 | if r.remote != nil { |
| 16 | out, err := r.remote.call(nil, remoteRequest{Method: "UpsertTimerRegistration", T: t, Expected: expected}, nil) |
| 17 | return out.Timer, err |
| 18 | } |
| 19 | |
| 20 | tx, err := r.sql.Begin() |
| 21 | if err != nil { |
| 22 | return nil, err |
| 23 | } |
| 24 | defer tx.Rollback() |
| 25 | old, err := scanTimer(tx.QueryRow(`SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, t.Host, t.Name)) |
| 26 | if errors.Is(err, sql.ErrNoRows) { |
| 27 | if expected != "" { |
| 28 | return nil, ErrTimerChanged |
| 29 | } |
| 30 | } else if err != nil { |
| 31 | return nil, err |
| 32 | } else { |
| 33 | if old.Kind != t.Kind { |
| 34 | return nil, ErrTimerKind |
| 35 | } |
| 36 | if expected == "" || old.Registration != expected { |
| 37 | return nil, ErrTimerChanged |
| 38 | } |
| 39 | } |
| 40 | rt, err := scanRoute(tx.QueryRow(`SELECT `+routeCols+` FROM routes WHERE host=?`, t.Host).Scan) |
| 41 | if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 42 | return nil, err |
| 43 | } |
| 44 | if err == nil && rt.ClaimOwnership.Valid() { |
| 45 | if t.ClaimOwner != "" && (t.ClaimIncarnation != rt.ClaimIncarnation || t.ClaimStamp != rt.ClaimStamp || t.ClaimOwner != rt.ClaimOwner) { |
| 46 | return nil, ErrClaimOwnership |
| 47 | } |
| 48 | t.ClaimOwnership = rt.ClaimOwnership |
| 49 | } |
| 50 | var id [16]byte |
| 51 | if _, err = rand.Read(id[:]); err != nil { |
| 52 | return nil, err |
| 53 | } |
| 54 | t.Registration = hex.EncodeToString(id[:]) |
| 55 | if t.ClaimOwnership.Valid() { |
| 56 | t.ClaimRowID = t.Registration |
| 57 | } |
| 58 | if err = upsertTimer(tx, t); err != nil { |
| 59 | return nil, err |
| 60 | } |
| 61 | row, err := scanTimer(tx.QueryRow(`SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, t.Host, t.Name)) |
| 62 | if err != nil { |
| 63 | return nil, err |
| 64 | } |
| 65 | if err := tx.Commit(); err != nil { |
| 66 | return nil, err |
| 67 | } |
| 68 | return &row, nil |
| 69 | } |
| 70 | |
| 71 | func (r *Registry) SetTimerStatusRegistration(host, name, registration, status string) error { |
| 72 | if r.remote != nil { |
| 73 | _, err := r.remote.call(nil, remoteRequest{Method: "SetTimerStatusRegistration", Host: host, Name: name, Registration: registration, Status: status}, nil) |
| 74 | return err |
| 75 | } |
| 76 | |
| 77 | if registration == "" { |
| 78 | return ErrTimerChanged |
| 79 | } |
| 80 | res, err := r.sql.Exec(`UPDATE timers SET last_status=? WHERE host=? AND name=? AND registration=?`, status, host, name, registration) |
| 81 | return timerRegistrationResult(res, err) |
| 82 | } |
| 83 | |
| 84 | func (r *Registry) FireTimerNowRegistration(host, name, registration string, now int64) error { |
| 85 | if r.remote != nil { |
| 86 | _, err := r.remote.call(nil, remoteRequest{Method: "FireTimerNowRegistration", Host: host, Name: name, Registration: registration, Now: now}, nil) |
| 87 | return err |
| 88 | } |
| 89 | |
| 90 | if registration == "" { |
| 91 | return ErrTimerChanged |
| 92 | } |
| 93 | var id [16]byte |
| 94 | if _, err := rand.Read(id[:]); err != nil { |
| 95 | return err |
| 96 | } |
| 97 | next := hex.EncodeToString(id[:]) |
| 98 | res, err := r.sql.Exec(`UPDATE timers SET next_at=?, due_at=?, attempts=0, registration=?, claim_row_id=CASE WHEN claim_owner='' THEN claim_row_id ELSE ? END WHERE host=? AND name=? AND registration=?`, now, now, next, next, host, name, registration) |
| 99 | return timerRegistrationResult(res, err) |
| 100 | } |
| 101 | |
| 102 | func timerRegistrationResult(res sql.Result, err error) error { |
| 103 | if err != nil { |
| 104 | return err |
| 105 | } |
| 106 | n, err := res.RowsAffected() |
| 107 | if err == nil && n == 0 { |
| 108 | return ErrTimerChanged |
| 109 | } |
| 110 | return err |
| 111 | } |
| 112 | |
amadan