Amortize hosted relay authorization and defer peer formation

docs/changes/unreleased/2162-hosted-relay-authority-fastpath.md

@@ -0,0 +1,23 @@
+---
+kind: fixed
+title: Reuse confirmed hosted authority and prepare new identity peers asynchronously
+pr: 2162
+surface: [remote, chat]
+invalidates:
+  - Hosted reads previously repeated quorum work and synchronous presence updates; they now reuse confirmed authority while failing closed when it expires.
+  - New empty identities previously waited for three voting replicas; their first change now commits on one voter before background joint-consensus formation.
+---
+
+Directory mutations remain consensus commits. Committed revocations close device
+watches and held store polls immediately. Authorization is checked once before
+blob work and repeated after I/O only when the applied index changed. Advisory
+presence is coalesced outside the request path.
+
+Internal quorum reads preserve an in-progress idle transition; ordinary reads
+still wake the group. Link approvals validate certificates and grants before
+reserving a pairing request. Exact retries of a committed decision remain valid
+after its original pending deadline; expired undecided requests remain refused.
+
+Existing identities retain their voting configuration. New singleton births
+persist immutable provenance that older relay code cannot serve; upgrades must
+preserve compatible recovery rather than restore a local directory writer.

internal/clusterstate/apply.go

@@ -150,7 +150,7 @@ func Sign(c Command, key ed25519.PrivateKey) (Command, error) {
 // immutable records from unread records.
 func Dependencies(c Command) []RecordKey {
 	keys := []RecordKey{QuiescenceKey(), IdentityKey(), DeviceKey(c.Actor), AuthorityKey(c.Actor), SessionKey(c.Actor, c.Session), SessionsKey()}
-	if c.Change.InitMembership != nil {
+	if c.Change.InitMembership != nil || c.Change.PrepareQuiescence != nil {
 		keys = append(keys, LegacyImportKey())
 	}
 	m := c.Change
@@ -360,11 +360,13 @@ func evaluateCommand(c Command, meta ApplicationMeta, read ReadRecord, paged boo
 	// Old imported groups retain their exact manifest and rows. Their FIRST
 	// registry command may use only a manifest-provisioned anchor key; it logs
 	// the authority installation instead of rewriting a frozen import snapshot.
-	if rows[AuthorityKey(c.Actor)] == nil && c.Change.InitMembership != nil {
+	// Before that installation, the initial singleton anchor may only prepare
+	// a quorum-confirmed idle marker for its actual one-voter configuration.
+	if rows[AuthorityKey(c.Actor)] == nil && (c.Change.InitMembership != nil || c.Change.PrepareQuiescence != nil && len(meta.Configuration.Voters) == 1 && !meta.Configuration.Joint()) {
 		var manifest LegacyManifest
 		if raw := rows[LegacyImportKey()]; raw != nil && Decode(raw, &manifest) == nil && manifest.Validate() == nil && manifest.Config.Cluster == c.Cluster && bytes.Equal(manifest.Config.SigningKey, identity.SigningKey) {
 			for i, node := range manifest.Config.Anchors {
-				if c.Actor == AnchorActor(node) && i < len(manifest.Config.AnchorKeys) {
+				if c.Actor == AnchorActor(node) && i < len(manifest.Config.AnchorKeys) && (c.Change.InitMembership != nil || i == 0 && meta.Configuration.Voters[0] == node) {
 					auth = Authority{Cluster: c.Cluster, Key: manifest.Config.AnchorKeys[i], Anchor: true}
 				}
 			}
@@ -419,7 +421,7 @@ func evaluateCommand(c Command, meta ApplicationMeta, read ReadRecord, paged boo
 			return out, nil
 		}
 	}
-	service := c.Change.RenewHolds != nil || c.Change.RecordUpload != nil || c.Change.AdmitJoin != nil || c.Change.CompleteJoin != nil || auth.Effects && (c.Change.WorkerBeatBatch != nil || c.Change.DeviceSeen != nil)
+	service := c.Change.RenewHolds != nil || c.Change.RecordUpload != nil || c.Change.AdmitJoin != nil || c.Change.CompleteJoin != nil || auth.Effects && (c.Change.WorkerBeatBatch != nil || c.Change.DeviceSeen != nil || c.Change.AdvanceTime != nil)
 	if service != auth.Effects || auth.Personal && auth.Effects || auth.Anchor && (auth.Personal || auth.Effects || (c.Change.EnrollMember == nil || c.Change.EnrollMember.Admission == nil) && (c.Change.RemoveMember == nil || c.Change.RemoveMember.Admission == nil) && c.Change.InitMembership == nil && c.Change.PrepareQuiescence == nil && c.Change.SettleMembership == nil && c.Change.SettleLeadership == nil && c.Change.CompleteLeadership == nil && c.Change.RestoreAnchors == nil && c.Change.CloseSession == nil) {
 		return out, nil
 	}
@@ -521,7 +523,10 @@ func evaluateCommand(c Command, meta ApplicationMeta, read ReadRecord, paged boo
 				}
 			}
 		} else if service {
-			code = applyEffect(c, out.Now, rows, put)
+			// A clock witness advances logical time without stamping presence.
+			if c.Change.AdvanceTime == nil {
+				code = applyEffect(c, out.Now, rows, put)
+			}
 		} else {
 			if assignmentMutation(c.Change) {
 				code = applyAssignment(c, out.Now, identity, rows, put, &out.Result)

internal/clusterstate/effects.go

@@ -147,7 +147,7 @@ func applyEffect(c Command, now int64, rows map[RecordKey][]byte, put func(Recor
 		return OK
 	case c.Change.AdmitJoin != nil:
 		v := c.Change.AdmitJoin
-		if v.Code == "" || len(v.PublicKey) != ed25519.PublicKeySize || v.Expires <= now || v.Expires > int64(MaxSafeInteger) {
+		if v.Code == "" || len(v.PublicKey) != ed25519.PublicKeySize || v.Expires > int64(MaxSafeInteger) {
 			return BadRequest
 		}
 		k := JoinKey(v.Code)
@@ -156,11 +156,19 @@ func applyEffect(c Command, now int64, rows map[RecordKey][]byte, put func(Recor
 			if Decode(raw, &r) != nil {
 				return BadRequest
 			}
-			if bytes.Equal(r.PublicKey, v.PublicKey) && r.Expires == v.Expires && r.Platform == PlatformOf(v.Platform) {
+			sameAdmission := bytes.Equal(r.PublicKey, v.PublicKey) && r.Expires == v.Expires && r.Platform == PlatformOf(v.Platform)
+			// Pending expiry cannot invalidate an already committed decision replay.
+			if v.Expires <= now && (!sameAdmission || r.Decision == nil) {
+				return BadRequest
+			}
+			if sameAdmission {
 				return OK
 			}
 			return Conflict
 		}
+		if v.Expires <= now {
+			return BadRequest
+		}
 		put(k, JoinRequest{Device: DeviceIDOfKey(v.PublicKey), PublicKey: v.PublicKey, Expires: v.Expires, Platform: PlatformOf(v.Platform)})
 		return OK
 	case c.Change.CompleteJoin != nil:

internal/clusterstate/effects_test.go

@@ -129,6 +129,63 @@ func TestR16R17R21EffectsAuthorizationCampaign(t *testing.T) {
 		})
 	}
 }
+func TestR16DecidedJoinAdmissionSurvivesPendingExpiry(t *testing.T) {
+	public := actorKey("joiner").Public().(ed25519.PublicKey)
+	admission := cs.AdmitJoin{Code: "request", PublicKey: public, Expires: 1002}
+	decision := cs.JoinDecision{Code: "request", Device: cs.DeviceIDOfKey(public), Cert: joinCert(t, "joiner"), Grant: "c2VhbGVk", Record: cs.Device{V: 1, Name: "c2VhbGVk"}}
+	t.Run("decided replay preserves the original decision", func(t *testing.T) {
+		a := newApplication(t)
+		if r := service(a, 1000, cs.Mutation{AdmitJoin: &admission}); r.Code != cs.OK {
+			t.Fatal(r)
+		}
+		if r := a.run("a", 1001, cs.Mutation{DecideJoin: &decision}); r.Code != cs.OK {
+			t.Fatal(r)
+		}
+		if r := service(a, 1003, cs.Mutation{AdmitJoin: &admission}); r.Code != cs.OK {
+			t.Fatal("pending expiry blocked an existing decision", r)
+		}
+		if r := a.run("a", 1004, cs.Mutation{DecideJoin: &decision}); r.Code != cs.OK {
+			t.Fatal("the original decision was not idempotent", r)
+		}
+		if r := a.run("a", 1005, cs.Mutation{DecideJoin: &cs.JoinDecision{Code: "request", Deny: true}}); r.Code != cs.AlreadyDecided {
+			t.Fatal("a contradictory decision changed the outcome", r)
+		}
+		extended := admission
+		extended.Expires = 2000
+		if r := service(a, 1006, cs.Mutation{AdmitJoin: &extended}); r.Code != cs.Conflict {
+			t.Fatal("re-admission changed a decided request's lifetime", r)
+		}
+		mismatched := admission
+		mismatched.PublicKey = actorKey("different").Public().(ed25519.PublicKey)
+		if r := service(a, 1007, cs.Mutation{AdmitJoin: &mismatched}); r.Code != cs.BadRequest {
+			t.Fatal("expired re-admission changed the immutable requester", r)
+		}
+	})
+	t.Run("expired pending admission cannot be revived", func(t *testing.T) {
+		a := newApplication(t)
+		if r := service(a, 1000, cs.Mutation{AdmitJoin: &admission}); r.Code != cs.OK {
+			t.Fatal(r)
+		}
+		if r := service(a, 1003, cs.Mutation{AdmitJoin: &admission}); r.Code != cs.BadRequest {
+			t.Fatal("an expired pending request was admitted again", r)
+		}
+		extended := admission
+		extended.Expires = 2000
+		if r := service(a, 1004, cs.Mutation{AdmitJoin: &extended}); r.Code != cs.Conflict {
+			t.Fatal("re-admission extended the pending lifetime", r)
+		}
+		if r := a.run("a", 1005, cs.Mutation{DecideJoin: &decision}); r.Code != cs.RequestGone {
+			t.Fatal("an expired pending request acquired a decision", r)
+		}
+	})
+	t.Run("unseen expired admission is refused", func(t *testing.T) {
+		a := newApplication(t)
+		if r := service(a, 1003, cs.Mutation{AdmitJoin: &admission}); r.Code != cs.BadRequest {
+			t.Fatal("an unseen expired request was admitted", r)
+		}
+	})
+}
+
 func TestR16PairingCertificateAndStickyRevocation(t *testing.T) {
 	a := newApplication(t)
 	pub := actorKey("joiner").Public().(ed25519.PublicKey)

internal/manual/chat/coordination-and-workers.md

@@ -100,6 +100,17 @@ An open Home or compatibility events viewer pings every 10 seconds. The platform
 
 Connections naming held chats share a renewal cycle every 30 seconds on both relays while fresh hold evidence is available. Each cycle uses bounded batches; the 10-second presence pings between cycles do not commit another renewal. Presence alone never renews a chat.
 
+## First connection and hosted read authority
+
+A new, empty identity can save its first change before the hosted relay finishes
+preparing its other copies. Machine coordination can briefly ask you to retry
+until that preparation finishes. Existing identities keep their voting group.
+
+Ordinary hosted reads reuse recently confirmed authority. Once that confirmation
+expires, an unavailable majority makes reads fail rather than use stale device
+permissions. Revocation and rotation remain committed changes; device activity
+is advisory and can appear after a short delay.
+
 ## Hosted relay memory capacity and retry
 
 After every voter has completed the group's idle barrier, the hosted relay frees its working memory for other households and keeps the saved directory in persistent storage. The next request reloads those copies and obtains a fresh majority response. Busy groups keep their memory. When busy groups fill the relay's memory allowance, another activation receives HTTP 503 with retry guidance; retrying after a group sleeps can succeed.

internal/raftabi/abi.go

@@ -206,6 +206,8 @@ func (r *Runtime) call(op uint32, d *decoder) ([]byte, error) {
 		return v1Commands(d)
 	case 41, 42, 43, 44, 45:
 		return r.pagedApplication(op, d)
+	case 49:
+		return r.formation(d)
 	case 39:
 		return hostedForward(d)
 	case 36:

internal/raftabi/formation.go

@@ -0,0 +1,27 @@
+package raftabi
+
+import (
+	cs "github.com/Agent-Field/codeaf/internal/clusterstate"
+)
+
+func (r *Runtime) formation(d *decoder) ([]byte, error) {
+	node := r.nodes[d.u32()]
+	if node == nil {
+		return nil, errFrame
+	}
+	var manifest cs.LegacyManifest
+	raw, birth, registry := d.blob(), d.blob(), d.blob()
+	if d.end() != nil || cs.Decode(raw, &manifest) != nil || !validSingleton(manifest, birth) || len(registry) > 0 {
+		return nil, errFrame
+	}
+	plan, err := node.PlanFormation(manifest.Config.Anchors)
+	if err != nil {
+		return nil, err
+	}
+	e := encoder{}
+	e.blob(plan.Change)
+	e.flag(plan.Checkpoint)
+	e.flag(plan.Settle)
+	e.flag(plan.Configure)
+	return e.b, nil
+}

internal/raftabi/hosted.go

@@ -22,7 +22,7 @@ func hostedEffect(d *decoder) ([]byte, error) {
 	if cs.Decode(raw, &change) != nil && (j.Decode(&change) != nil || j.Decode(new(any)) != io.EOF) {
 		return nil, errFrame
 	}
-	if change.DeviceSeen == nil && change.RenewHolds == nil && change.AdmitJoin == nil && change.CompleteJoin == nil && change.RecordUpload == nil {
+	if change.AdvanceTime == nil && change.DeviceSeen == nil && change.RenewHolds == nil && change.AdmitJoin == nil && change.CompleteJoin == nil && change.RecordUpload == nil {
 		return nil, errFrame
 	}
 	key := ed25519.NewKeyFromSeed(seed)

internal/raftabi/legacy.go

@@ -4,6 +4,7 @@ import (
 	"github.com/Agent-Field/codeaf/internal/clusterstate"
 	pb "go.etcd.io/raft/v3/raftpb"
 	"google.golang.org/protobuf/proto"
+	"strconv"
 )
 
 // The hosted source feeds original SQL JSON to this bounded conversion call.
@@ -36,6 +37,10 @@ func legacyCodec(d *decoder) ([]byte, error) {
 func legacyGenesis(d *decoder) ([]byte, error) {
 	var m clusterstate.LegacyManifest
 	raw := d.blob()
+	var birth []byte
+	if len(d.b) > 0 {
+		birth = d.blob()
+	}
 	if e := d.end(); e != nil {
 		return nil, e
 	}
@@ -45,7 +50,14 @@ func legacyGenesis(d *decoder) ([]byte, error) {
 	e := encoder{}
 	e.message(&pb.Entry{Index: proto.Uint64(1), Term: proto.Uint64(1)})
 	e.message(&pb.HardState{Term: proto.Uint64(1), Commit: proto.Uint64(1)})
-	e.message(&pb.ConfState{Voters: m.Config.Anchors})
+	voters := m.Config.Anchors
+	if len(birth) > 0 {
+		if !validSingleton(m, birth) {
+			return nil, clusterstate.ErrMigration
+		}
+		voters = voters[:1]
+	}
+	e.message(&pb.ConfState{Voters: voters})
 	return e.b, nil
 }
 
@@ -73,3 +85,16 @@ func legacyAuthorities(d *decoder) ([]byte, error) {
 	}
 	return out.b, nil
 }
+
+func validSingleton(m clusterstate.LegacyManifest, raw []byte) bool {
+	var birth struct {
+		Cluster string   `json:"cluster"`
+		Digest  string   `json:"digest"`
+		Voters  []string `json:"voters"`
+	}
+	if clusterstate.Decode(raw, &birth) != nil || m.Validate() != nil || m.Version != 0 || len(m.Config.Anchors) != 3 || birth.Cluster != m.Config.Cluster || birth.Digest != m.Digest || len(birth.Voters) != 1 || birth.Voters[0] != strconv.FormatUint(m.Config.Anchors[0], 10) {
+		return false
+	}
+	rows, err := clusterstate.LegacyAuthorities(m.Config)
+	return err == nil && m.Rows == uint64(len(rows)+1)
+}

internal/raftabi/membership.go

@@ -48,6 +48,16 @@ func learnerReady(d *decoder) ([]byte, error) {
 
 func anchorRegistry(d *decoder) ([]byte, error) {
 	raw := d.blob()
+	var conf *pb.ConfState
+	if len(d.b) > 0 {
+		conf = new(pb.ConfState)
+		d.message(conf)
+		var manifest clusterstate.LegacyManifest
+		manifestRaw, birth := d.blob(), d.blob()
+		if clusterstate.Decode(manifestRaw, &manifest) != nil || !validSingleton(manifest, birth) {
+			return nil, errFrame
+		}
+	}
 	if err := d.end(); err != nil {
 		return nil, err
 	}
@@ -62,6 +72,23 @@ func anchorRegistry(d *decoder) ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
+	if conf != nil {
+		for _, node := range append(append(append([]uint64{}, conf.GetVoters()...), conf.GetVotersOutgoing()...), conf.GetLearners()...) {
+			found := false
+			for _, peer := range peers {
+				found = found || peer.Node == node
+			}
+			if !found {
+				return nil, errFrame
+			}
+		}
+		for i := range peers {
+			peers[i].Voter = false
+			for _, node := range append(append([]uint64{}, conf.GetVoters()...), conf.GetVotersOutgoing()...) {
+				peers[i].Voter = peers[i].Voter || peers[i].Node == node
+			}
+		}
+	}
 	b, err := clusterstate.Encode(peers)
 	if err != nil {
 		return nil, err

internal/raftcore/core.go

@@ -258,14 +258,10 @@ func (c *Core) Step(input Input) (effects Effects, err error) {
 			effects.SnapshotNeeded = c.SnapshotNeeded()
 		}
 	}()
+	idleRead := false
 	if idle, ok := input.(IdleRead); ok {
-		active, sleeping := c.schedule.active, c.schedule.sleeping
+		idleRead = true
 		input = Read{RequestID: idle.RequestID}
-		defer func() {
-			if err == nil && !sleeping {
-				c.schedule.active = active
-			}
-		}()
 	}
 	if c.stopped {
 		return effects, ErrStopped
@@ -315,7 +311,11 @@ func (c *Core) Step(input Input) (effects Effects, err error) {
 		return effects, ErrBusy
 	}
 	switch input.(type) {
-	case Read, Propose, Configure, Transfer:
+	case Read:
+		if !idleRead || c.schedule.sleeping {
+			c.wakeScheduling()
+		}
+	case Propose, Configure, Transfer:
 		c.wakeScheduling()
 	case PeerMessage:
 		if c.schedule.sleeping {