update

pkg/actor/grain_pool.go

@@ -19,8 +19,13 @@ type GrainPool[V any] interface {
 	OwnerHost(id uint64) (Host[V], bool)
 	Hostname() string
 	TakeOwnership(id uint64)
-	HandleOwnershipChange(host string, ids []uint64) error
-	HandleRemoteExpiry(host string, ids []uint64) error
+	// HandleOwnershipChange processes a remote first-touch ownership
+	// claim. lastChanges is the parallel []int64 of UnixNano stamps from
+	// the announcer; -1 entries fall back to the pre-arbitration
+	// "always accept remote" behaviour so mixed-version rollouts
+	// remain safe.
+	HandleOwnershipChange(host string, ids []uint64, lastChanges []int64) error
+	HandleRemoteExpiry(host string, ids []uint64, lastChanges []int64) error
 	Negotiate(otherHosts []string)
 	GetLocalIds() []uint64
 	IsHealthy() bool
@@ -32,7 +37,13 @@ type GrainPool[V any] interface {
 
 // Host abstracts a remote node capable of proxying cart requests.
 type Host[V any] interface {
-	AnnounceExpiry(ids []uint64)
+	// AnnounceExpiry broadcasts per-id eviction decisions. lastChanges is
+	// a parallel []int64 of UnixNano stamps taken at the moment the
+	// grain was dropped from the local cache; it is informational on
+	// this path (the receiver just removes the id from its remoteOwners
+	// map) but kept in the wire signature for symmetry with
+	// AnnounceOwnership.
+	AnnounceExpiry(ids []uint64, lastChanges []int64)
 	Negotiate(otherHosts []string) ([]string, error)
 	Name() string
 	Proxy(id uint64, w http.ResponseWriter, r *http.Request, customBody io.Reader) (bool, error)
@@ -42,5 +53,14 @@ type Host[V any] interface {
 	Close() error
 	Ping() bool
 	IsHealthy() bool
-	AnnounceOwnership(ownerHost string, ids []uint64)
+	// AnnounceOwnership broadcasts a first-touch ownership claim.
+	// lastChanges is a parallel []int64 of UnixNano stamps of the
+	// announcing pod's local grain's GetLastChange() at the moment of
+	// broadcast; receivers use it as the first-spawn-wins oracle for
+	// concurrent cold-cache first-touch (smaller stamp = older spawn =
+	// owns the grain; on equal stamps the lexicographically smaller
+	// hostname wins). A stamp of -1 means "no local grain to back this
+	// id" — receivers treat that as legacy/no-arbitration and accept
+	// the remote claim as before.
+	AnnounceOwnership(ownerHost string, ids []uint64, lastChanges []int64)
 }

pkg/actor/grpc_server.go

@@ -88,7 +88,7 @@ func (s *ControlServer[V]) AnnounceOwnership(ctx context.Context, req *messages.
 	)
 	announceOwnershipCalls.Add(ctx, 1, metric.WithAttributes(attribute.String("host", req.Host)))
 
-	err := s.pool.HandleOwnershipChange(req.Host, req.Ids)
+	err := s.pool.HandleOwnershipChange(req.Host, req.Ids, req.LastChanges)
 	if err != nil {
 		span.RecordError(err)
 		return &messages.OwnerChangeAck{
@@ -138,7 +138,7 @@ func (s *ControlServer[V]) AnnounceExpiry(ctx context.Context, req *messages.Exp
 	)
 	announceExpiryCalls.Add(ctx, 1, metric.WithAttributes(attribute.String("host", req.Host)))
 
-	err := s.pool.HandleRemoteExpiry(req.Host, req.Ids)
+	err := s.pool.HandleRemoteExpiry(req.Host, req.Ids, req.LastChanges)
 	if err != nil {
 		span.RecordError(err)
 	}

pkg/actor/grpc_server_test.go

@@ -41,8 +41,8 @@ func (m *mockGrainPool) TakeOwnership(id uint64) {}
 
 func (m *mockGrainPool) Hostname() string { return "test-host" }
 
-func (m *mockGrainPool) HandleOwnershipChange(host string, ids []uint64) error { return nil }
-func (m *mockGrainPool) HandleRemoteExpiry(host string, ids []uint64) error    { return nil }
+func (m *mockGrainPool) HandleOwnershipChange(host string, ids []uint64, _ []int64) error { return nil }
+func (m *mockGrainPool) HandleRemoteExpiry(host string, ids []uint64, _ []int64) error    { return nil }
 func (m *mockGrainPool) Negotiate(hosts []string)                              {}
 func (m *mockGrainPool) GetLocalIds() []uint64                                 { return []uint64{} }
 func (m *mockGrainPool) RemoveHost(host string)                                {}

pkg/actor/simple_grain_pool.go

@@ -45,6 +45,16 @@ type SimpleGrainPool[V any] struct {
 	purgeTicker *time.Ticker
 }
 
+// noLocalStamp is the sentinel value used in the parallel []int64
+// lastChanges slice to mark an id for which the broadcaster has no
+// local grain (e.g. TakeOwnership called before anyone has loaded the
+// grain, or a custom Host implementation that doesn't carry a stamp).
+// Receivers treat this sentinel as "no arbitration possible" and fall
+// back to the pre-arbitration "always accept remote" verdict so the
+// behaviour is identical to running an older binary that didn't carry
+// stamps at all.
+const noLocalStamp int64 = -1
+
 type GrainPoolConfig[V any] struct {
 	Hostname         string
 	Spawn            func(ctx context.Context, id uint64) (Grain[V], error)
@@ -114,17 +124,20 @@ func (p *SimpleGrainPool[V]) RemoveListener(listener LogListener) {
 func (p *SimpleGrainPool[V]) purge() {
 	purgeLimit := time.Now().Add(-p.ttl)
 	purgedIds := make([]uint64, 0, len(p.grains))
+	purgedStamps := make([]int64, 0, len(p.grains))
 	p.localMu.Lock()
 	evicted := 0
 	for id, grain := range p.grains {
 		if grain.GetLastAccess().Before(purgeLimit) {
-			purgedIds = append(purgedIds, id)
 			if p.destroy != nil {
 				if err := p.destroy(grain); err != nil {
 					log.Printf("failed to destroy grain %d: %v", id, err)
 				}
 			}
-
+			// Capture the lastChange stamp BEFORE delete so peers
+			// receive an honest per-id eviction record on the wire.
+			purgedIds = append(purgedIds, id)
+			purgedStamps = append(purgedStamps, grain.GetLastChange().UnixNano())
 			delete(p.grains, id)
 			evicted++
 		}
@@ -135,7 +148,7 @@ func (p *SimpleGrainPool[V]) purge() {
 		p.metrics.SetPoolStats(remaining, p.poolSize)
 	}
 	p.forAllHosts(func(remote Host[V]) {
-		remote.AnnounceExpiry(purgedIds)
+		remote.AnnounceExpiry(purgedIds, purgedStamps)
 	})
 
 }
@@ -161,7 +174,14 @@ func (p *SimpleGrainPool[V]) GetLocalIds() []uint64 {
 	return ids
 }
 
-func (p *SimpleGrainPool[V]) HandleRemoteExpiry(host string, ids []uint64) error {
+func (p *SimpleGrainPool[V]) HandleRemoteExpiry(host string, ids []uint64, _ []int64) error {
+	// lastChanges is informational on this path; expiry is unilateral
+	// from the announcer's perspective — we just drop the id from any
+	// remoteOwners mapping so future cross-pod forwards stop hitting a
+	// dead peer. The parallel stamp slice is kept in the signature for
+	// wire symmetry with AnnounceOwnership and so a future hardening
+	// pass can use stamps to detect ghost-evictions (e.g. peer evicted
+	// with a stamp newer than our last seen Apply result).
 	p.remoteMu.Lock()
 	defer p.remoteMu.Unlock()
 	for _, id := range ids {
@@ -170,7 +190,7 @@ func (p *SimpleGrainPool[V]) HandleRemoteExpiry(host string, ids []uint64) error
 	return nil
 }
 
-func (p *SimpleGrainPool[V]) HandleOwnershipChange(host string, ids []uint64) error {
+func (p *SimpleGrainPool[V]) HandleOwnershipChange(host string, ids []uint64, lastChanges []int64) error {
 	p.remoteMu.RLock()
 	remoteHost, exists := p.remoteHosts[host]
 	p.remoteMu.RUnlock()
@@ -186,15 +206,71 @@ func (p *SimpleGrainPool[V]) HandleOwnershipChange(host string, ids []uint64) er
 	// p.localMu holder.
 	p.remoteMu.Lock()
 	p.localMu.Lock()
-	for _, id := range ids {
-		log.Printf("Handling ownership change for cart %d to host %s", id, host)
+	var accepted, kept, legacy int
+	for i, id := range ids {
+		// remoteStamp is the announcer's lastChange at broadcast time,
+		// or noLocalStamp (-1) if the announcer didn't have a local
+		// grain (legacy / TakeOwnership path). The
+		// 0..len(lastChanges)-1 slice is indexed parallel to ids; out
+		// of range falls back to noLocalStamp (= "no arbitration").
+		var remoteStamp int64 = noLocalStamp
+		if i < len(lastChanges) {
+			remoteStamp = lastChanges[i]
+		}
+		var localStamp int64 = noLocalStamp
+		var hasLocal bool
+		if g, ok := p.grains[id]; ok {
+			localStamp = g.GetLastChange().UnixNano()
+			hasLocal = true
+		}
+
+		// First-spawn-wins arbitration. Five distinct cases, ordered:
+		//   1) remoteStamp == noLocalStamp  → announcer didn't carry a
+		//      real stamp (legacy or TakeOwnership). Keep current
+		//      "always accept remote" behaviour so mixed-version
+		//      rollouts and unbroadcast-advertised TakeOwnership don't
+		//      regress.
+		//   2) localStamp  == noLocalStamp  → we have no grain at all.
+		//      Accept remote so the next read forwards to the right
+		//      owner instead of us spawning a parallel projection.
+		//   3) stamps differ → first-spawn-wins. The pod that spawned
+		//      the grain earlier (smaller stamp) owns it; its broadcast
+		//      is authoritative. The newer pod's projection is
+		//      redundant — we drop our local copy and defer to remote
+		//      so future OwnerHost lookups route correctly.
+		//   4) stamps equal  → deterministic tie-break: lexicographically
+		//      smaller hostname wins (both pods compute the same
+		//      verdict, so the cold-cache first-touch cannot flip-flop).
+		var keepLocal bool
+		switch {
+		case remoteStamp == noLocalStamp:
+			keepLocal = false
+			if hasLocal {
+				legacy++
+			}
+		case localStamp == noLocalStamp:
+			keepLocal = false
+		case localStamp != remoteStamp:
+			keepLocal = localStamp < remoteStamp
+		default:
+			keepLocal = p.hostname < host
+		}
+		if keepLocal {
+			kept++
+			continue
+		}
 		delete(p.grains, id)
 		p.remoteOwners[id] = remoteHost
+		accepted++
 	}
 	remaining := len(p.grains)
 	p.localMu.Unlock()
 	p.remoteMu.Unlock()
 	p.metrics.SetPoolStats(remaining, p.poolSize)
+	if kept > 0 || accepted > 0 || legacy > 0 {
+		log.Printf("HandleOwnershipChange from %s: accepted_remote=%d kept_local=%d legacy_no_stamp=%d",
+			host, accepted, kept, legacy)
+	}
 	return nil
 }
 
@@ -418,8 +494,27 @@ func (p *SimpleGrainPool[V]) broadcastOwnership(ids []uint64) {
 		return
 	}
 
+	// Capture each id's grain.lastChange UnixNano stamp under the local
+	// read-lock so peers receive an honest oracle for the
+	// first-spawn-wins arbitration. For ids that don't have a local
+	// grain (e.g. TakeOwnership called before anyone has loaded the
+	// grain, or a caller pushing ids into the pool from outside the
+	// spawn path), substitute the noLocalStamp sentinel — receivers see
+	// -1 and fall back to "always accept remote", preserving the
+	// pre-arbitration semantics for that case.
+	p.localMu.RLock()
+	stamps := make([]int64, 0, len(ids))
+	for _, id := range ids {
+		if g, ok := p.grains[id]; ok {
+			stamps = append(stamps, g.GetLastChange().UnixNano())
+		} else {
+			stamps = append(stamps, noLocalStamp)
+		}
+	}
+	p.localMu.RUnlock()
+
 	p.forAllHosts(func(rh Host[V]) {
-		rh.AnnounceOwnership(p.hostname, ids)
+		rh.AnnounceOwnership(p.hostname, ids, stamps)
 	})
 	log.Printf("%s taking ownership of %d ids", p.hostname, len(ids))
 	// go p.statsUpdate()

pkg/actor/simple_grain_pool_test.go

@@ -45,7 +45,7 @@ func newMockHost(name string) *mockHost {
 	return &mockHost{name: name, healthy: true}
 }
 
-func (m *mockHost) AnnounceExpiry(ids []uint64) {
+func (m *mockHost) AnnounceExpiry(ids []uint64, _ []int64) {
 	m.mu.Lock()
 	m.expiry = append(m.expiry, ids...)
 	m.mu.Unlock()
@@ -84,7 +84,7 @@ func (m *mockHost) Ping() bool { return m.healthy }
 
 func (m *mockHost) IsHealthy() bool { return m.healthy }
 
-func (m *mockHost) AnnounceOwnership(_ string, ids []uint64) {
+func (m *mockHost) AnnounceOwnership(_ string, ids []uint64, _ []int64) {
 	m.mu.Lock()
 	m.ownership = append(m.ownership, ids...)
 	m.mu.Unlock()
@@ -247,11 +247,19 @@ func TestSimpleGrainPoolHandleOwnershipChange(t *testing.T) {
 	host := newMockHost("peer-d")
 	pool := newTestPool(t, func(string) (Host[testState], error) { return host, nil })
 
+	// First-spawn-wins arbitration: a remote broadcast wins only when its
+	// lastChange stamp is strictly earlier than the local grain's stamp
+	// (or the local grain is absent). Capture the local spawn time once
+	// and pass a stamp 1ns earlier to the remote so the test continues to
+	// assert "external owner takes over given an earlier remote spawn"
+	// under the new arbitration rule.
+	spawnTime := time.Now()
 	pool.localMu.Lock()
-	pool.grains[7] = &testGrain{id: 7, accessed: time.Now()}
+	pool.grains[7] = &testGrain{id: 7, accessed: spawnTime}
 	pool.localMu.Unlock()
 
-	if err := pool.HandleOwnershipChange(host.name, []uint64{7}); err != nil {
+	remoteStamp := spawnTime.UnixNano() - 1
+	if err := pool.HandleOwnershipChange(host.name, []uint64{7}, []int64{remoteStamp}); err != nil {
 		t.Fatal(err)
 	}
 
@@ -268,6 +276,121 @@ func TestSimpleGrainPoolHandleOwnershipChange(t *testing.T) {
 	}
 }
 
+// TestSimpleGrainPoolHandleOwnershipChangeArbitration verifies the
+// first-spawn-wins arbitration rules for HandleOwnershipChange:
+//
+//   - Remote with earlier lastChange than local  → accept remote.
+//   - Remote with later lastChange than local   → keep local.
+//   - Equal lastChange                           → lower hostname wins.
+//   - Remote stamp == -1 (legacy path)           → accept remote (pre-arbitration).
+//   - Local grain absent                        → accept remote.
+func TestSimpleGrainPoolHandleOwnershipChangeArbitration(t *testing.T) {
+	host := newMockHost("peer-arbitrate")
+	// Each table case constructs a fresh pool inside t.Run — we don't
+	// share state across cases because per-case hostname differs
+	// (hostname is the tie-break oracle on equal lastChange stamps).
+
+	now := time.Now().UnixNano()
+
+	cases := []struct {
+		name          string
+		name_local    string // pool hostname for ordering check
+		id            uint64
+		localStamp    int64  // UnixNano of the local grain; -1 = no local grain
+		remoteStamp   int64  // UnixNano stamped on the broadcast
+		expectLocal   bool   // true → local grain still resident
+		expectRemote  bool   // true → remoteOwners[id] = host
+	}{
+		{
+			name:        "remote older stamp wins",
+			name_local:  "zzz", // pool above "peer-arbitrate" lex
+			id:          10,
+			localStamp:  now + 100,
+			remoteStamp: now,
+			expectLocal: false,
+			expectRemote: true,
+		},
+		{
+			name:        "local earlier stamp keeps",
+			name_local:  "aaa", // pool below "peer-arbitrate" lex but stamp decides
+			id:          11,
+			localStamp:  now,
+			remoteStamp: now + 100,
+			expectLocal: true,
+			expectRemote: false,
+		},
+		{
+			name:        "equal stamp + lower local hostname keeps",
+			name_local:  "a-peer", // below host
+			id:          12,
+			localStamp:  now,
+			remoteStamp: now,
+			expectLocal: true,
+			expectRemote: false,
+		},
+		{
+			name:        "equal stamp + higher local hostname defers",
+			name_local:  "zzz-peer", // above host
+			id:          13,
+			localStamp:  now,
+			remoteStamp: now,
+			expectLocal: false,
+			expectRemote: true,
+		},
+		{
+			name:        "legacy (no remote stamp) accepts remote",
+			name_local:  "zzz", // we'd otherwise keep local
+			id:          14,
+			localStamp:  now + 100,
+			remoteStamp: -1,
+			expectLocal: false,
+			expectRemote: true,
+		},
+		{
+			name:        "no local grain accepts remote",
+			name_local:  "aaa",
+			id:          15,
+			localStamp:  -1,
+			remoteStamp: now,
+			expectLocal: false,
+			expectRemote: true,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Fresh pool per case so hostname differences don't leak.
+			p := newTestPool(t, func(string) (Host[testState], error) { return host, nil })
+			p.hostname = tc.name_local
+
+			p.localMu.Lock()
+			if tc.localStamp != -1 {
+				p.grains[tc.id] = &testGrain{id: tc.id, accessed: time.Unix(0, tc.localStamp)}
+			}
+			p.localMu.Unlock()
+
+			if err := p.HandleOwnershipChange(host.name, []uint64{tc.id}, []int64{tc.remoteStamp}); err != nil {
+				t.Fatal(err)
+			}
+
+			p.localMu.RLock()
+			_, hasLocal := p.grains[tc.id]
+			p.localMu.RUnlock()
+			if hasLocal != tc.expectLocal {
+				t.Fatalf("local grain presence = %v, want %v", hasLocal, tc.expectLocal)
+			}
+
+			owner, ok := p.OwnerHost(tc.id)
+			if ok != tc.expectRemote {
+				t.Fatalf("OwnerHost ok = %v, want %v", ok, tc.expectRemote)
+			}
+			if tc.expectRemote && (owner == nil || owner.Name() != host.name) {
+				t.Fatalf("Owner = %v, want host %q", owner, host.name)
+			}
+		})
+	}
+}
+
 func TestSimpleGrainPoolPurgeEvictsStaleGrains(t *testing.T) {
 	pool := newTestPool(t, func(string) (Host[testState], error) {
 		return nil, fmt.Errorf("no remotes")

pkg/proxy/remotehost.go

@@ -263,10 +263,16 @@ func (h *RemoteHost[V]) GetActorIds() []uint64 {
 	return reply.GetIds()
 }
 
-func (h *RemoteHost[V]) AnnounceOwnership(ownerHost string, uids []uint64) {
+func (h *RemoteHost[V]) AnnounceOwnership(ownerHost string, uids []uint64, lastChanges []int64) {
+	// lastChanges is parallel to uids and carries each grain's
+	// UnixNano lastChange stamp at the moment of broadcast. Receivers
+	// use it for first-spawn-wins arbitration on concurrent cold
+	// cache first touch; -1 entries signal "no local grain to back
+	// this id" and trigger the legacy "always accept remote" fallback.
 	_, err := h.controlClient.AnnounceOwnership(context.Background(), &messages.OwnershipAnnounce{
-		Host: ownerHost,
-		Ids:  uids,
+		Host:        ownerHost,
+		Ids:         uids,
+		LastChanges: lastChanges,
 	})
 	if err != nil {
 		log.Printf("ownership announce to %s failed: %v", h.host, err)
@@ -276,10 +282,14 @@ func (h *RemoteHost[V]) AnnounceOwnership(ownerHost string, uids []uint64) {
 	h.missedPings = 0
 }
 
-func (h *RemoteHost[V]) AnnounceExpiry(uids []uint64) {
+func (h *RemoteHost[V]) AnnounceExpiry(uids []uint64, lastChanges []int64) {
+	// lastChanges is a parallel UnixNano stamp slice; informational on
+	// this path (expiry is unilateral from the broadcaster), kept on
+	// the wire for symmetry with AnnounceOwnership.
 	_, err := h.controlClient.AnnounceExpiry(context.Background(), &messages.ExpiryAnnounce{
-		Host: h.host,
-		Ids:  uids,
+		Host:        h.host,
+		Ids:         uids,
+		LastChanges: lastChanges,
 	})
 	if err != nil {
 		log.Printf("expiry announce to %s failed: %v", h.host, err)