amadan

carlos / platform Public

relay: authority record, serving lease, heartbeat (spec §3, §5)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev fb6c07572e3228b5026e1f93339b5cc1e32175bd parent df42101
3 files changed, +144 −0
  • internal/relay/authority.go +83 −0
  • internal/relay/authority_test.go +46 −0
  • internal/relay/memstore_test.go +15 −0
diff --git a/internal/relay/authority.go b/internal/relay/authority.go
new file mode 100644
index 000000000..0c96456cc
--- /dev/null
+++ b/internal/relay/authority.go
@@ -0,0 +1,83 @@
+package relay
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const (
+ AuthorityKey = "control/relay.json"
+ LeaseMaxAge = 15 * time.Minute
+)
+
+type AuthorityApp struct {
+ Account string `json:"account"`
+ App string `json:"app"`
+ Incarnation string `json:"incarnation"`
+ Regions []string `json:"regions"`
+}
+
+type Authority struct {
+ Apps []AuthorityApp `json:"apps"`
+ WrittenAt time.Time `json:"written_at"`
+}
+
+func (a *Authority) Lookup(account, app string) (AuthorityApp, bool) {
+ if a == nil {
+ return AuthorityApp{}, false
+ }
+ for _, x := range a.Apps {
+ if x.Account == account && x.App == app {
+ return x, true
+ }
+ }
+ return AuthorityApp{}, false
+}
+
+func ReadAuthority(ctx context.Context, s release.Store) (*Authority, string, error) {
+ body, etag, err := s.Get(ctx, AuthorityKey)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, "", nil
+ }
+ if err != nil {
+ return nil, "", err
+ }
+ var a Authority
+ if err := json.Unmarshal(body, &a); err != nil {
+ return nil, etag, err
+ }
+ return &a, etag, nil
+}
+
+// Lease is the relay's serving lease: live while the last SUCCESSFUL
+// authority read is younger than LeaseMaxAge. A failed read renews nothing.
+type Lease struct{ last time.Time }
+
+func (l *Lease) Renew(now time.Time) { l.last = now }
+func (l *Lease) Live(now time.Time) bool { return !l.last.IsZero() && now.Sub(l.last) <= LeaseMaxAge }
+
+func HeartbeatKey(box string) string { return "control/relay-ack/" + box + ".json" }
+
+type Heartbeat struct {
+ Box string `json:"box"`
+ Region string `json:"region"`
+ WrittenAt time.Time `json:"written_at"` // of the authority record held; zero if none
+ ReadAt time.Time `json:"read_at"`
+ Listeners []string `json:"listeners"`
+ Kids []string `json:"kids"`
+ MastersError string `json:"masters_error,omitempty"`
+}
+
+func WriteHeartbeat(ctx context.Context, s release.Store, hb Heartbeat) error {
+ body, err := json.Marshal(hb)
+ if err != nil {
+ return err
+ }
+ _, err = s.Put(ctx, HeartbeatKey(hb.Box), body, "")
+ return err
+}
diff --git a/internal/relay/authority_test.go b/internal/relay/authority_test.go
new file mode 100644
index 000000000..34310f48a
--- /dev/null
+++ b/internal/relay/authority_test.go
@@ -0,0 +1,46 @@
+package relay
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "time"
+)
+
+func TestAuthorityRoundTrip(t *testing.T) {
+ s := newMemStore(t)
+ ctx := context.Background()
+ a, _, err := ReadAuthority(ctx, s)
+ if err != nil || a != nil {
+ t.Fatalf("absent = %v %v, want nil,nil", a, err)
+ }
+ body, _ := json.Marshal(Authority{Apps: []AuthorityApp{{Account: "abc", App: "meet", Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1"}}}, WrittenAt: time.Unix(1_800_000_000, 0).UTC()})
+ if _, err := s.Put(ctx, AuthorityKey, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ a, etag, err := ReadAuthority(ctx, s)
+ if err != nil || etag == "" {
+ t.Fatal(err)
+ }
+ if got, ok := a.Lookup("abc", "meet"); !ok || got.Incarnation != "0123456789abcdef" {
+ t.Fatalf("lookup = %+v %v", got, ok)
+ }
+ if _, ok := a.Lookup("abc", "other"); ok {
+ t.Fatal("unknown app found")
+ }
+}
+
+func TestLease(t *testing.T) {
+ var l Lease
+ t0 := time.Unix(1_800_000_000, 0)
+ if l.Live(t0) {
+ t.Fatal("never renewed should not be live")
+ }
+ l.Renew(t0)
+ if !l.Live(t0.Add(14 * time.Minute)) {
+ t.Fatal("14m should be live")
+ }
+ if l.Live(t0.Add(16 * time.Minute)) {
+ t.Fatal("16m should not be live")
+ }
+}
diff --git a/internal/relay/memstore_test.go b/internal/relay/memstore_test.go
new file mode 100644
index 000000000..f5d1bc73b
--- /dev/null
+++ b/internal/relay/memstore_test.go
@@ -0,0 +1,15 @@
+package relay
+
+import (
+ "testing"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// newMemStore is the release.Store every test in this package uses: a
+// directory store under t.TempDir(), so CAS (Put with an etag) behaves
+// exactly as it does against the deployment bucket.
+func newMemStore(t *testing.T) release.Store {
+ t.Helper()
+ return release.NewDirStore(t.TempDir())
+}