| 1 | package relay |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/release" |
| 11 | ) |
| 12 | |
| 13 | const ( |
| 14 | AuthorityKey = "control/relay.json" |
| 15 | LeaseMaxAge = 15 * time.Minute |
| 16 | ) |
| 17 | |
| 18 | type AuthorityApp struct { |
| 19 | Account string `json:"account"` |
| 20 | App string `json:"app"` |
| 21 | Incarnation string `json:"incarnation"` |
| 22 | Regions []string `json:"regions"` |
| 23 | } |
| 24 | |
| 25 | type Authority struct { |
| 26 | Apps []AuthorityApp `json:"apps"` |
| 27 | WrittenAt time.Time `json:"written_at"` |
| 28 | } |
| 29 | |
| 30 | func (a *Authority) Lookup(account, app string) (AuthorityApp, bool) { |
| 31 | if a == nil { |
| 32 | return AuthorityApp{}, false |
| 33 | } |
| 34 | for _, x := range a.Apps { |
| 35 | if x.Account == account && x.App == app { |
| 36 | return x, true |
| 37 | } |
| 38 | } |
| 39 | return AuthorityApp{}, false |
| 40 | } |
| 41 | |
| 42 | func ReadAuthority(ctx context.Context, s release.Store) (*Authority, string, error) { |
| 43 | body, etag, err := s.Get(ctx, AuthorityKey) |
| 44 | if errors.Is(err, os.ErrNotExist) { |
| 45 | return nil, "", nil |
| 46 | } |
| 47 | if err != nil { |
| 48 | return nil, "", err |
| 49 | } |
| 50 | var a Authority |
| 51 | if err := json.Unmarshal(body, &a); err != nil { |
| 52 | return nil, etag, err |
| 53 | } |
| 54 | return &a, etag, nil |
| 55 | } |
| 56 | |
| 57 | // Lease is the relay's serving lease: live while the last SUCCESSFUL |
| 58 | // authority read is younger than LeaseMaxAge. A failed read renews nothing. |
| 59 | type Lease struct{ last time.Time } |
| 60 | |
| 61 | func (l *Lease) Renew(now time.Time) { l.last = now } |
| 62 | func (l *Lease) Live(now time.Time) bool { return !l.last.IsZero() && now.Sub(l.last) <= LeaseMaxAge } |
| 63 | |
| 64 | // Last is the time of the last successful read, zero if there has never |
| 65 | // been one. It is what the heartbeat reports as read_at: the wall clock |
| 66 | // would say "just now" on a relay whose every read has failed for twenty |
| 67 | // minutes, which is the one thing a console watching this record must not |
| 68 | // be told. |
| 69 | func (l *Lease) Last() time.Time { return l.last } |
| 70 | |
| 71 | func HeartbeatKey(box string) string { return "control/relay-ack/" + box + ".json" } |
| 72 | |
| 73 | type Heartbeat struct { |
| 74 | Box string `json:"box"` |
| 75 | Region string `json:"region"` |
| 76 | WrittenAt time.Time `json:"written_at"` // of the authority record held; zero if none |
| 77 | // ReadAt is the last SUCCESSFUL authority read — the lease clock, not |
| 78 | // the wall clock — so the gap between it and the heartbeat's own |
| 79 | // arrival is how far the relay has drifted from the record it serves. |
| 80 | ReadAt time.Time `json:"read_at"` |
| 81 | Listeners []string `json:"listeners"` |
| 82 | Kids []string `json:"kids"` |
| 83 | // Retiring is kid -> retire_at for the loaded masters that carry one, |
| 84 | // so `carlos relay status` can tell an operator mid-rotation when the |
| 85 | // old key may leave the box's masters file (discussion #120). Additive: |
| 86 | // a relay older than this field simply reports none. |
| 87 | Retiring map[string]time.Time `json:"retiring,omitempty"` |
| 88 | // The two "why is this record stale" fields: the last masters-file |
| 89 | // reload error and the last authority read error, either of which |
| 90 | // explains a ReadAt that has stopped moving. |
| 91 | MastersError string `json:"masters_error,omitempty"` |
| 92 | AuthorityError string `json:"authority_error,omitempty"` |
| 93 | } |
| 94 | |
| 95 | // WriteHeartbeat writes a box's heartbeat by read-then-CAS: Get the current |
| 96 | // etag (absent means create with ifMatch ""), then Put with it. A box owns |
| 97 | // its own heartbeat key, so an ErrCASConflict can only be our own previous |
| 98 | // write racing a restart — retry the whole sequence once more and return |
| 99 | // whatever the second attempt reports. |
| 100 | func WriteHeartbeat(ctx context.Context, s release.Store, hb Heartbeat) error { |
| 101 | body, err := json.Marshal(hb) |
| 102 | if err != nil { |
| 103 | return err |
| 104 | } |
| 105 | key := HeartbeatKey(hb.Box) |
| 106 | var lastErr error |
| 107 | for attempt := 0; attempt < 2; attempt++ { |
| 108 | etag := "" |
| 109 | _, curEtag, err := s.Get(ctx, key) |
| 110 | switch { |
| 111 | case errors.Is(err, os.ErrNotExist): |
| 112 | // no existing record: create-only Put below. |
| 113 | case err != nil: |
| 114 | return err |
| 115 | default: |
| 116 | etag = curEtag |
| 117 | } |
| 118 | if _, err := s.Put(ctx, key, body, etag); err != nil { |
| 119 | lastErr = err |
| 120 | if errors.Is(err, release.ErrCASConflict) { |
| 121 | continue |
| 122 | } |
| 123 | return err |
| 124 | } |
| 125 | return nil |
| 126 | } |
| 127 | return lastErr |
| 128 | } |
| 129 | |
amadan