diff --git a/cmd/profile/clustered_applier.go b/cmd/profile/clustered_applier.go new file mode 100644 index 0000000..c204bda --- /dev/null +++ b/cmd/profile/clustered_applier.go @@ -0,0 +1,63 @@ +package main + +import ( + "context" + + "google.golang.org/protobuf/proto" + + "git.k6n.net/mats/go-cart-actor/pkg/actor" + "git.k6n.net/mats/go-cart-actor/pkg/profile" +) + +// clusterAwareApplier is a ProfileApplier that chooses between an +// in-process pool call and a remote forward based on which replica +// currently owns the grain. It is the single seam at which the auth +// server (`customerauth.Server`) and the UCP customer handler +// (`ucp.CustomerServer`) ever speak to the grain pool — keeping the +// decision of "go to the authoritative owner" vs "spawn locally" in +// one place avoids the split-brain hazard introduced when multiple +// code paths (HTTP middleware + the handlers' own pool.Get) could each +// independently decide to spawn a stale or empty grain and broadcast +// conflicting ownership for it. +// +// The decision rule: +// +// - pool.OwnerHost(id) returns a remote host → forward Get/Apply +// to that host. The host holds the authoritative in-memory state. +// - pool.OwnerHost(id) returns no host → delegate to +// pool.Get / pool.Apply. On a local cache miss pool.Get spawns +// the grain from disk on this pod, broadcasts the new ownership, +// and returns the grain; this is the only code path that ever +// claims a grain on this pod. +// +// pool.Get's spawn path is unchanged on purpose: the disk-backed event +// log is the source of truth, and the local cache is just a projection +// rebuilt from it. The risk surface that the applicr layer removes is +// the HTTP-middleware fall-through that re-entered the same pool.Get +// from a different code path and duplicated the decision. +type clusterAwareApplier struct { + pool *actor.SimpleGrainPool[profile.ProfileGrain] +} + +// Get returns the current state of grain id, preferring the +// authoritative remote owner when one is registered. With no remote +// owner the pool spawns the grain locally (from the disk-backed event +// log), takes ownership, and returns the grain. +func (a *clusterAwareApplier) Get(ctx context.Context, id uint64) (*profile.ProfileGrain, error) { + if owner, ok := a.pool.OwnerHost(id); ok { + return owner.Get(ctx, id) + } + return a.pool.Get(ctx, id) +} + +// Apply mutates grain id with messages, forwarding to the authoritative +// remote owner when one is registered. With no remote owner the pool +// spawns the grain locally, applies the mutation, takes ownership, and +// returns the mutated state — its listeners (including the AMQP +// mutation feed) fire from this pod. +func (a *clusterAwareApplier) Apply(ctx context.Context, id uint64, msgs ...proto.Message) (*actor.MutationResult[profile.ProfileGrain], error) { + if owner, ok := a.pool.OwnerHost(id); ok { + return owner.Apply(ctx, id, msgs...) + } + return a.pool.Apply(ctx, id, msgs...) +} diff --git a/cmd/profile/main.go b/cmd/profile/main.go index eaa3e2c..4828e5b 100644 --- a/cmd/profile/main.go +++ b/cmd/profile/main.go @@ -243,12 +243,28 @@ func main() { // UCP Customer REST adapter auditLogPath := filepath.Join(profileDir, "audit.log") - customerUCP := ucp.CustomerHandler(pool, auditLogPath, + // Session signer is shared by the auth server (HMAC verifier for the + // "sid" cookie); NewSigner is stateless once built so a single + // instance can be reused across the auth server's NewServer call. + sessionSigner := customerauth.NewSigner(authSecret()) + + // Cluster-aware ProfileApplier: routes Get/Apply to the replica that + // currently owns the grain. With a remote owner we forward there so + // the request reads the authoritative in-memory state; with no owner + // we let pool.Get / pool.Apply spawn the grain locally from the + // disk-backed event log and broadcast ownership. Centralising this + // choice in the applicr (rather than HTTP middleware) prevents the + // split-brain hazard where a non-owner pod reads a stale or empty + // grain from disk and caches it under its own (conflicting) + // ownership. + applier := &clusterAwareApplier{pool: pool} + + customerUCP := ucp.CustomerHandler(applier, auditLogPath, ucp.WithEmailIndex(emailIndex), ucp.WithCredentialDeleter(credStore), ) - if signer := loadUCPProfileSigner(); signer != nil { - customerUCP = ucp.WithSigning(customerUCP, signer) + if ucpSigner := loadUCPProfileSigner(); ucpSigner != nil { + customerUCP = ucp.WithSigning(customerUCP, ucpSigner) log.Print("ucp customer signing enabled") } // StripPrefix is required because the sub-mux (CustomerHandler) registers @@ -258,7 +274,7 @@ func main() { // Customer auth: password signup/login + session cookies + identity linking. authNotifier := selectAuthNotifier() - authHandler := customerauth.NewServer(credStore, pool, customerauth.NewSigner(authSecret()), 0, customerauth.Options{ + authHandler := customerauth.NewServer(credStore, applier, sessionSigner, 0, customerauth.Options{ Notifier: authNotifier, Limiter: limiter, BaseURL: os.Getenv("CUSTOMER_AUTH_BASE_URL"), diff --git a/pkg/actor/grain_pool.go b/pkg/actor/grain_pool.go index 2928f0a..1c1660a 100644 --- a/pkg/actor/grain_pool.go +++ b/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) } diff --git a/pkg/actor/grpc_server.go b/pkg/actor/grpc_server.go index 20b50f7..0bad97f 100644 --- a/pkg/actor/grpc_server.go +++ b/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) } diff --git a/pkg/actor/grpc_server_test.go b/pkg/actor/grpc_server_test.go index 76e8b1d..643ad33 100644 --- a/pkg/actor/grpc_server_test.go +++ b/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) {} diff --git a/pkg/actor/simple_grain_pool.go b/pkg/actor/simple_grain_pool.go index 5edb117..95f9d89 100644 --- a/pkg/actor/simple_grain_pool.go +++ b/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() diff --git a/pkg/actor/simple_grain_pool_test.go b/pkg/actor/simple_grain_pool_test.go index aea330a..133a8b4 100644 --- a/pkg/actor/simple_grain_pool_test.go +++ b/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") diff --git a/pkg/proxy/remotehost.go b/pkg/proxy/remotehost.go index ef21aa5..6f36127 100644 --- a/pkg/proxy/remotehost.go +++ b/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) diff --git a/proto/control/control_plane.pb.go b/proto/control/control_plane.pb.go index fbcbad4..e6f920e 100644 --- a/proto/control/control_plane.pb.go +++ b/proto/control/control_plane.pb.go @@ -346,11 +346,20 @@ func (x *ClosingNotice) GetHost() string { } // OwnershipAnnounce broadcasts first-touch ownership claims for cart IDs. -// First claim wins; receivers SHOULD NOT overwrite an existing different owner. +// Receivers arbitrate concurrent cold-cache first-touch claims using +// last_changes: the pod whose local grain has the earliest lastChange +// wins; ties resolve to the lexicographically smaller hostname. type OwnershipAnnounce struct { - state protoimpl.MessageState `protogen:"open.v1"` - Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"` // announcing host - Ids []uint64 `protobuf:"varint,2,rep,packed,name=ids,proto3" json:"ids,omitempty"` // newly claimed cart ids + state protoimpl.MessageState `protogen:"open.v1"` + Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"` // announcing host + Ids []uint64 `protobuf:"varint,2,rep,packed,name=ids,proto3" json:"ids,omitempty"` // newly claimed cart ids + // last_changes is the per-id UnixNano timestamp of the announcing + // pod's local grain's lastChange at the moment of broadcast. -1 + // means the announcer didn't have a local grain for this id (legacy + // path / TakeOwnership without prior load); receivers treat -1 as + // "no arbitration done" and accept the remote claim as before. The + // slice is aligned 1:1 with ids. + LastChanges []int64 `protobuf:"varint,3,rep,packed,name=last_changes,json=lastChanges,proto3" json:"last_changes,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -399,11 +408,21 @@ func (x *OwnershipAnnounce) GetIds() []uint64 { return nil } +func (x *OwnershipAnnounce) GetLastChanges() []int64 { + if x != nil { + return x.LastChanges + } + return nil +} + // ExpiryAnnounce broadcasts that a host evicted the provided cart IDs. +// last_changes is carried for wire-shape symmetry with OwnershipAnnounce +// and is informational here (expiry is unilateral). type ExpiryAnnounce struct { state protoimpl.MessageState `protogen:"open.v1"` Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"` Ids []uint64 `protobuf:"varint,2,rep,packed,name=ids,proto3" json:"ids,omitempty"` + LastChanges []int64 `protobuf:"varint,3,rep,packed,name=last_changes,json=lastChanges,proto3" json:"last_changes,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -452,6 +471,13 @@ func (x *ExpiryAnnounce) GetIds() []uint64 { return nil } +func (x *ExpiryAnnounce) GetLastChanges() []int64 { + if x != nil { + return x.LastChanges + } + return nil +} + type ApplyRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Id uint64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` @@ -730,91 +756,96 @@ var file_control_plane_proto_rawDesc = string([]byte{ 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x23, 0x0a, 0x0d, 0x43, 0x6c, 0x6f, 0x73, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x74, 0x69, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, - 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x22, 0x39, 0x0a, 0x11, + 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x22, 0x5c, 0x0a, 0x11, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x73, 0x68, 0x69, 0x70, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x69, 0x64, 0x73, 0x18, 0x02, 0x20, 0x03, - 0x28, 0x04, 0x52, 0x03, 0x69, 0x64, 0x73, 0x22, 0x36, 0x0a, 0x0e, 0x45, 0x78, 0x70, 0x69, 0x72, - 0x79, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, 0x73, - 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x12, 0x10, 0x0a, - 0x03, 0x69, 0x64, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x04, 0x52, 0x03, 0x69, 0x64, 0x73, 0x22, - 0x50, 0x0a, 0x0c, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, - 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x02, 0x69, 0x64, 0x12, - 0x30, 0x0a, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, - 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, - 0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, - 0x73, 0x22, 0x1c, 0x0a, 0x0a, 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, - 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x02, 0x69, 0x64, 0x22, - 0x36, 0x0a, 0x08, 0x47, 0x65, 0x74, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x2a, 0x0a, 0x05, 0x67, - 0x72, 0x61, 0x69, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, - 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, - 0x52, 0x05, 0x67, 0x72, 0x61, 0x69, 0x6e, 0x22, 0x79, 0x0a, 0x0e, 0x4d, 0x75, 0x74, 0x61, 0x74, - 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, - 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x2e, 0x0a, - 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, - 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, - 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x19, 0x0a, - 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x05, - 0x65, 0x72, 0x72, 0x6f, 0x72, 0x88, 0x01, 0x01, 0x42, 0x08, 0x0a, 0x06, 0x5f, 0x65, 0x72, 0x72, - 0x6f, 0x72, 0x22, 0x7f, 0x0a, 0x0b, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x52, 0x65, 0x73, 0x75, 0x6c, - 0x74, 0x12, 0x2a, 0x0a, 0x05, 0x73, 0x74, 0x61, 0x74, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, + 0x28, 0x04, 0x52, 0x03, 0x69, 0x64, 0x73, 0x12, 0x21, 0x0a, 0x0c, 0x6c, 0x61, 0x73, 0x74, 0x5f, + 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x03, 0x52, 0x0b, 0x6c, + 0x61, 0x73, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x73, 0x22, 0x59, 0x0a, 0x0e, 0x45, 0x78, + 0x70, 0x69, 0x72, 0x79, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, + 0x68, 0x6f, 0x73, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, + 0x12, 0x10, 0x0a, 0x03, 0x69, 0x64, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x04, 0x52, 0x03, 0x69, + 0x64, 0x73, 0x12, 0x21, 0x0a, 0x0c, 0x6c, 0x61, 0x73, 0x74, 0x5f, 0x63, 0x68, 0x61, 0x6e, 0x67, + 0x65, 0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x03, 0x52, 0x0b, 0x6c, 0x61, 0x73, 0x74, 0x43, 0x68, + 0x61, 0x6e, 0x67, 0x65, 0x73, 0x22, 0x50, 0x0a, 0x0c, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x52, 0x65, + 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x04, 0x52, 0x02, 0x69, 0x64, 0x12, 0x30, 0x0a, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, + 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, + 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x08, 0x6d, + 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x22, 0x1c, 0x0a, 0x0a, 0x47, 0x65, 0x74, 0x52, 0x65, + 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x04, 0x52, 0x02, 0x69, 0x64, 0x22, 0x36, 0x0a, 0x08, 0x47, 0x65, 0x74, 0x52, 0x65, 0x70, 0x6c, + 0x79, 0x12, 0x2a, 0x0a, 0x05, 0x67, 0x72, 0x61, 0x69, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, - 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x05, 0x73, 0x74, 0x61, 0x74, 0x65, 0x12, 0x44, 0x0a, - 0x09, 0x6d, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, - 0x32, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, - 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4d, 0x75, 0x74, 0x61, 0x74, 0x69, - 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x09, 0x6d, 0x75, 0x74, 0x61, 0x74, 0x69, - 0x6f, 0x6e, 0x73, 0x32, 0xd6, 0x05, 0x0a, 0x0c, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x50, - 0x6c, 0x61, 0x6e, 0x65, 0x12, 0x48, 0x0a, 0x04, 0x50, 0x69, 0x6e, 0x67, 0x12, 0x1d, 0x2e, 0x63, - 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, - 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x21, 0x2e, 0x63, 0x6f, - 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, - 0x61, 0x67, 0x65, 0x73, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x5d, - 0x0a, 0x09, 0x4e, 0x65, 0x67, 0x6f, 0x74, 0x69, 0x61, 0x74, 0x65, 0x12, 0x28, 0x2e, 0x63, 0x6f, + 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x05, 0x67, 0x72, 0x61, 0x69, 0x6e, 0x22, 0x79, 0x0a, + 0x0e, 0x4d, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, + 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, + 0x79, 0x70, 0x65, 0x12, 0x2e, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, + 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, + 0x61, 0x67, 0x65, 0x12, 0x19, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x03, 0x20, 0x01, + 0x28, 0x09, 0x48, 0x00, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x88, 0x01, 0x01, 0x42, 0x08, + 0x0a, 0x06, 0x5f, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x22, 0x7f, 0x0a, 0x0b, 0x41, 0x70, 0x70, 0x6c, + 0x79, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x2a, 0x0a, 0x05, 0x73, 0x74, 0x61, 0x74, 0x65, + 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x05, 0x73, 0x74, + 0x61, 0x74, 0x65, 0x12, 0x44, 0x0a, 0x09, 0x6d, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, + 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, + 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, + 0x4d, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x09, + 0x6d, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x32, 0xd6, 0x05, 0x0a, 0x0c, 0x43, 0x6f, + 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x50, 0x6c, 0x61, 0x6e, 0x65, 0x12, 0x48, 0x0a, 0x04, 0x50, 0x69, + 0x6e, 0x67, 0x12, 0x1d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, + 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x45, 0x6d, 0x70, 0x74, + 0x79, 0x1a, 0x21, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, + 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, + 0x65, 0x70, 0x6c, 0x79, 0x12, 0x5d, 0x0a, 0x09, 0x4e, 0x65, 0x67, 0x6f, 0x74, 0x69, 0x61, 0x74, + 0x65, 0x12, 0x28, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, + 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4e, 0x65, 0x67, 0x6f, 0x74, + 0x69, 0x61, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4e, 0x65, 0x67, 0x6f, 0x74, 0x69, 0x61, 0x74, 0x65, 0x52, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, - 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4e, - 0x65, 0x67, 0x6f, 0x74, 0x69, 0x61, 0x74, 0x65, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x58, 0x0a, - 0x10, 0x47, 0x65, 0x74, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x41, 0x63, 0x74, 0x6f, 0x72, 0x49, 0x64, - 0x73, 0x12, 0x1d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, - 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, - 0x1a, 0x25, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, - 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x41, 0x63, 0x74, 0x6f, 0x72, 0x49, - 0x64, 0x73, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x66, 0x0a, 0x11, 0x41, 0x6e, 0x6e, 0x6f, 0x75, - 0x6e, 0x63, 0x65, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x73, 0x68, 0x69, 0x70, 0x12, 0x29, 0x2e, 0x63, - 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, - 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x73, 0x68, 0x69, 0x70, 0x41, - 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x1a, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, + 0x70, 0x6c, 0x79, 0x12, 0x58, 0x0a, 0x10, 0x47, 0x65, 0x74, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x41, + 0x63, 0x74, 0x6f, 0x72, 0x49, 0x64, 0x73, 0x12, 0x1d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, - 0x2e, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x41, 0x63, 0x6b, 0x12, - 0x52, 0x0a, 0x05, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x12, 0x24, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, - 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, - 0x73, 0x2e, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, - 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, - 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x52, 0x65, 0x73, - 0x75, 0x6c, 0x74, 0x12, 0x60, 0x0a, 0x0e, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x45, - 0x78, 0x70, 0x69, 0x72, 0x79, 0x12, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, - 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x45, - 0x78, 0x70, 0x69, 0x72, 0x79, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x1a, 0x26, 0x2e, + 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x25, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, + 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, + 0x41, 0x63, 0x74, 0x6f, 0x72, 0x49, 0x64, 0x73, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x66, 0x0a, + 0x11, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x73, 0x68, + 0x69, 0x70, 0x12, 0x29, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, + 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4f, 0x77, 0x6e, 0x65, + 0x72, 0x73, 0x68, 0x69, 0x70, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x1a, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x43, 0x68, 0x61, 0x6e, - 0x67, 0x65, 0x41, 0x63, 0x6b, 0x12, 0x58, 0x0a, 0x07, 0x43, 0x6c, 0x6f, 0x73, 0x69, 0x6e, 0x67, - 0x12, 0x25, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, - 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x43, 0x6c, 0x6f, 0x73, 0x69, 0x6e, - 0x67, 0x4e, 0x6f, 0x74, 0x69, 0x63, 0x65, 0x1a, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, - 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, - 0x2e, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x41, 0x63, 0x6b, 0x12, - 0x4b, 0x0a, 0x03, 0x47, 0x65, 0x74, 0x12, 0x22, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, + 0x67, 0x65, 0x41, 0x63, 0x6b, 0x12, 0x52, 0x0a, 0x05, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x12, 0x24, + 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, + 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x52, 0x65, 0x71, + 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, + 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x41, 0x70, + 0x70, 0x6c, 0x79, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x60, 0x0a, 0x0e, 0x41, 0x6e, 0x6e, + 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x45, 0x78, 0x70, 0x69, 0x72, 0x79, 0x12, 0x26, 0x2e, 0x63, 0x6f, + 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, + 0x61, 0x67, 0x65, 0x73, 0x2e, 0x45, 0x78, 0x70, 0x69, 0x72, 0x79, 0x41, 0x6e, 0x6e, 0x6f, 0x75, + 0x6e, 0x63, 0x65, 0x1a, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, + 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4f, 0x77, 0x6e, + 0x65, 0x72, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x41, 0x63, 0x6b, 0x12, 0x58, 0x0a, 0x07, 0x43, + 0x6c, 0x6f, 0x73, 0x69, 0x6e, 0x67, 0x12, 0x25, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, - 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x63, 0x6f, 0x6e, - 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, - 0x67, 0x65, 0x73, 0x2e, 0x47, 0x65, 0x74, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x42, 0x45, 0x5a, 0x43, - 0x67, 0x69, 0x74, 0x2e, 0x6b, 0x36, 0x6e, 0x2e, 0x6e, 0x65, 0x74, 0x2f, 0x6d, 0x61, 0x74, 0x73, - 0x2f, 0x67, 0x6f, 0x2d, 0x63, 0x61, 0x72, 0x74, 0x2d, 0x61, 0x63, 0x74, 0x6f, 0x72, 0x2f, 0x70, - 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x3b, 0x63, 0x6f, 0x6e, - 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, - 0x67, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x43, 0x6c, 0x6f, 0x73, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x74, 0x69, 0x63, 0x65, 0x1a, 0x26, 0x2e, + 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, + 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x43, 0x68, 0x61, 0x6e, + 0x67, 0x65, 0x41, 0x63, 0x6b, 0x12, 0x4b, 0x0a, 0x03, 0x47, 0x65, 0x74, 0x12, 0x22, 0x2e, 0x63, + 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, + 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, + 0x1a, 0x20, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, + 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x47, 0x65, 0x74, 0x52, 0x65, 0x70, + 0x6c, 0x79, 0x42, 0x45, 0x5a, 0x43, 0x67, 0x69, 0x74, 0x2e, 0x6b, 0x36, 0x6e, 0x2e, 0x6e, 0x65, + 0x74, 0x2f, 0x6d, 0x61, 0x74, 0x73, 0x2f, 0x67, 0x6f, 0x2d, 0x63, 0x61, 0x72, 0x74, 0x2d, 0x61, + 0x63, 0x74, 0x6f, 0x72, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, + 0x6f, 0x6c, 0x3b, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, + 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, + 0x33, }) var ( diff --git a/proto/control_plane.pb.go b/proto/control_plane.pb.go new file mode 100644 index 0000000..e6f920e --- /dev/null +++ b/proto/control_plane.pb.go @@ -0,0 +1,933 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.5 +// protoc v7.35.1 +// source: control_plane.proto + +package control_plane_messages + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + anypb "google.golang.org/protobuf/types/known/anypb" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// Empty request placeholder (common pattern). +type Empty struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Empty) Reset() { + *x = Empty{} + mi := &file_control_plane_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Empty) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Empty) ProtoMessage() {} + +func (x *Empty) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Empty.ProtoReflect.Descriptor instead. +func (*Empty) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{0} +} + +// Ping reply includes responding host and its current unix time (seconds). +type PingReply struct { + state protoimpl.MessageState `protogen:"open.v1"` + Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"` + UnixTime int64 `protobuf:"varint,2,opt,name=unix_time,json=unixTime,proto3" json:"unix_time,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *PingReply) Reset() { + *x = PingReply{} + mi := &file_control_plane_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *PingReply) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PingReply) ProtoMessage() {} + +func (x *PingReply) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PingReply.ProtoReflect.Descriptor instead. +func (*PingReply) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{1} +} + +func (x *PingReply) GetHost() string { + if x != nil { + return x.Host + } + return "" +} + +func (x *PingReply) GetUnixTime() int64 { + if x != nil { + return x.UnixTime + } + return 0 +} + +// NegotiateRequest carries the caller's full view of known hosts (including self). +type NegotiateRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + KnownHosts []string `protobuf:"bytes,1,rep,name=known_hosts,json=knownHosts,proto3" json:"known_hosts,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *NegotiateRequest) Reset() { + *x = NegotiateRequest{} + mi := &file_control_plane_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *NegotiateRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*NegotiateRequest) ProtoMessage() {} + +func (x *NegotiateRequest) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use NegotiateRequest.ProtoReflect.Descriptor instead. +func (*NegotiateRequest) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{2} +} + +func (x *NegotiateRequest) GetKnownHosts() []string { + if x != nil { + return x.KnownHosts + } + return nil +} + +// NegotiateReply returns the callee's healthy hosts (including itself). +type NegotiateReply struct { + state protoimpl.MessageState `protogen:"open.v1"` + Hosts []string `protobuf:"bytes,1,rep,name=hosts,proto3" json:"hosts,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *NegotiateReply) Reset() { + *x = NegotiateReply{} + mi := &file_control_plane_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *NegotiateReply) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*NegotiateReply) ProtoMessage() {} + +func (x *NegotiateReply) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use NegotiateReply.ProtoReflect.Descriptor instead. +func (*NegotiateReply) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{3} +} + +func (x *NegotiateReply) GetHosts() []string { + if x != nil { + return x.Hosts + } + return nil +} + +// CartIdsReply returns the list of cart IDs (string form) currently owned locally. +type ActorIdsReply struct { + state protoimpl.MessageState `protogen:"open.v1"` + Ids []uint64 `protobuf:"varint,1,rep,packed,name=ids,proto3" json:"ids,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ActorIdsReply) Reset() { + *x = ActorIdsReply{} + mi := &file_control_plane_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ActorIdsReply) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ActorIdsReply) ProtoMessage() {} + +func (x *ActorIdsReply) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ActorIdsReply.ProtoReflect.Descriptor instead. +func (*ActorIdsReply) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{4} +} + +func (x *ActorIdsReply) GetIds() []uint64 { + if x != nil { + return x.Ids + } + return nil +} + +// OwnerChangeAck retained as response type for Closing RPC (ConfirmOwner removed). +type OwnerChangeAck struct { + state protoimpl.MessageState `protogen:"open.v1"` + Accepted bool `protobuf:"varint,1,opt,name=accepted,proto3" json:"accepted,omitempty"` + Message string `protobuf:"bytes,2,opt,name=message,proto3" json:"message,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *OwnerChangeAck) Reset() { + *x = OwnerChangeAck{} + mi := &file_control_plane_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *OwnerChangeAck) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*OwnerChangeAck) ProtoMessage() {} + +func (x *OwnerChangeAck) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use OwnerChangeAck.ProtoReflect.Descriptor instead. +func (*OwnerChangeAck) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{5} +} + +func (x *OwnerChangeAck) GetAccepted() bool { + if x != nil { + return x.Accepted + } + return false +} + +func (x *OwnerChangeAck) GetMessage() string { + if x != nil { + return x.Message + } + return "" +} + +// ClosingNotice notifies peers this host is terminating (so they can drop / re-resolve). +type ClosingNotice struct { + state protoimpl.MessageState `protogen:"open.v1"` + Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ClosingNotice) Reset() { + *x = ClosingNotice{} + mi := &file_control_plane_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ClosingNotice) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ClosingNotice) ProtoMessage() {} + +func (x *ClosingNotice) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ClosingNotice.ProtoReflect.Descriptor instead. +func (*ClosingNotice) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{6} +} + +func (x *ClosingNotice) GetHost() string { + if x != nil { + return x.Host + } + return "" +} + +// OwnershipAnnounce broadcasts first-touch ownership claims for cart IDs. +// Receivers arbitrate concurrent cold-cache first-touch claims using +// last_changes: the pod whose local grain has the earliest lastChange +// wins; ties resolve to the lexicographically smaller hostname. +type OwnershipAnnounce struct { + state protoimpl.MessageState `protogen:"open.v1"` + Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"` // announcing host + Ids []uint64 `protobuf:"varint,2,rep,packed,name=ids,proto3" json:"ids,omitempty"` // newly claimed cart ids + // last_changes is the per-id UnixNano timestamp of the announcing + // pod's local grain's lastChange at the moment of broadcast. -1 + // means the announcer didn't have a local grain for this id (legacy + // path / TakeOwnership without prior load); receivers treat -1 as + // "no arbitration done" and accept the remote claim as before. The + // slice is aligned 1:1 with ids. + LastChanges []int64 `protobuf:"varint,3,rep,packed,name=last_changes,json=lastChanges,proto3" json:"last_changes,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *OwnershipAnnounce) Reset() { + *x = OwnershipAnnounce{} + mi := &file_control_plane_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *OwnershipAnnounce) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*OwnershipAnnounce) ProtoMessage() {} + +func (x *OwnershipAnnounce) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[7] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use OwnershipAnnounce.ProtoReflect.Descriptor instead. +func (*OwnershipAnnounce) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{7} +} + +func (x *OwnershipAnnounce) GetHost() string { + if x != nil { + return x.Host + } + return "" +} + +func (x *OwnershipAnnounce) GetIds() []uint64 { + if x != nil { + return x.Ids + } + return nil +} + +func (x *OwnershipAnnounce) GetLastChanges() []int64 { + if x != nil { + return x.LastChanges + } + return nil +} + +// ExpiryAnnounce broadcasts that a host evicted the provided cart IDs. +// last_changes is carried for wire-shape symmetry with OwnershipAnnounce +// and is informational here (expiry is unilateral). +type ExpiryAnnounce struct { + state protoimpl.MessageState `protogen:"open.v1"` + Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"` + Ids []uint64 `protobuf:"varint,2,rep,packed,name=ids,proto3" json:"ids,omitempty"` + LastChanges []int64 `protobuf:"varint,3,rep,packed,name=last_changes,json=lastChanges,proto3" json:"last_changes,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExpiryAnnounce) Reset() { + *x = ExpiryAnnounce{} + mi := &file_control_plane_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExpiryAnnounce) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExpiryAnnounce) ProtoMessage() {} + +func (x *ExpiryAnnounce) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[8] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExpiryAnnounce.ProtoReflect.Descriptor instead. +func (*ExpiryAnnounce) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{8} +} + +func (x *ExpiryAnnounce) GetHost() string { + if x != nil { + return x.Host + } + return "" +} + +func (x *ExpiryAnnounce) GetIds() []uint64 { + if x != nil { + return x.Ids + } + return nil +} + +func (x *ExpiryAnnounce) GetLastChanges() []int64 { + if x != nil { + return x.LastChanges + } + return nil +} + +type ApplyRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Id uint64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` + Messages []*anypb.Any `protobuf:"bytes,2,rep,name=messages,proto3" json:"messages,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ApplyRequest) Reset() { + *x = ApplyRequest{} + mi := &file_control_plane_proto_msgTypes[9] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ApplyRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ApplyRequest) ProtoMessage() {} + +func (x *ApplyRequest) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[9] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ApplyRequest.ProtoReflect.Descriptor instead. +func (*ApplyRequest) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{9} +} + +func (x *ApplyRequest) GetId() uint64 { + if x != nil { + return x.Id + } + return 0 +} + +func (x *ApplyRequest) GetMessages() []*anypb.Any { + if x != nil { + return x.Messages + } + return nil +} + +type GetRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Id uint64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetRequest) Reset() { + *x = GetRequest{} + mi := &file_control_plane_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetRequest) ProtoMessage() {} + +func (x *GetRequest) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[10] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetRequest.ProtoReflect.Descriptor instead. +func (*GetRequest) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{10} +} + +func (x *GetRequest) GetId() uint64 { + if x != nil { + return x.Id + } + return 0 +} + +type GetReply struct { + state protoimpl.MessageState `protogen:"open.v1"` + Grain *anypb.Any `protobuf:"bytes,1,opt,name=grain,proto3" json:"grain,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetReply) Reset() { + *x = GetReply{} + mi := &file_control_plane_proto_msgTypes[11] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetReply) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetReply) ProtoMessage() {} + +func (x *GetReply) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[11] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetReply.ProtoReflect.Descriptor instead. +func (*GetReply) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{11} +} + +func (x *GetReply) GetGrain() *anypb.Any { + if x != nil { + return x.Grain + } + return nil +} + +type MutationResult struct { + state protoimpl.MessageState `protogen:"open.v1"` + Type string `protobuf:"bytes,1,opt,name=type,proto3" json:"type,omitempty"` + Message *anypb.Any `protobuf:"bytes,2,opt,name=message,proto3" json:"message,omitempty"` + Error *string `protobuf:"bytes,3,opt,name=error,proto3,oneof" json:"error,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *MutationResult) Reset() { + *x = MutationResult{} + mi := &file_control_plane_proto_msgTypes[12] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *MutationResult) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*MutationResult) ProtoMessage() {} + +func (x *MutationResult) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[12] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use MutationResult.ProtoReflect.Descriptor instead. +func (*MutationResult) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{12} +} + +func (x *MutationResult) GetType() string { + if x != nil { + return x.Type + } + return "" +} + +func (x *MutationResult) GetMessage() *anypb.Any { + if x != nil { + return x.Message + } + return nil +} + +func (x *MutationResult) GetError() string { + if x != nil && x.Error != nil { + return *x.Error + } + return "" +} + +type ApplyResult struct { + state protoimpl.MessageState `protogen:"open.v1"` + State *anypb.Any `protobuf:"bytes,1,opt,name=state,proto3" json:"state,omitempty"` + Mutations []*MutationResult `protobuf:"bytes,2,rep,name=mutations,proto3" json:"mutations,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ApplyResult) Reset() { + *x = ApplyResult{} + mi := &file_control_plane_proto_msgTypes[13] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ApplyResult) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ApplyResult) ProtoMessage() {} + +func (x *ApplyResult) ProtoReflect() protoreflect.Message { + mi := &file_control_plane_proto_msgTypes[13] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ApplyResult.ProtoReflect.Descriptor instead. +func (*ApplyResult) Descriptor() ([]byte, []int) { + return file_control_plane_proto_rawDescGZIP(), []int{13} +} + +func (x *ApplyResult) GetState() *anypb.Any { + if x != nil { + return x.State + } + return nil +} + +func (x *ApplyResult) GetMutations() []*MutationResult { + if x != nil { + return x.Mutations + } + return nil +} + +var File_control_plane_proto protoreflect.FileDescriptor + +var file_control_plane_proto_rawDesc = string([]byte{ + 0x0a, 0x13, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x2e, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x16, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, + 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x1a, 0x19, 0x67, + 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x61, + 0x6e, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x07, 0x0a, 0x05, 0x45, 0x6d, 0x70, 0x74, + 0x79, 0x22, 0x3c, 0x0a, 0x09, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x12, + 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, + 0x73, 0x74, 0x12, 0x1b, 0x0a, 0x09, 0x75, 0x6e, 0x69, 0x78, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, + 0x02, 0x20, 0x01, 0x28, 0x03, 0x52, 0x08, 0x75, 0x6e, 0x69, 0x78, 0x54, 0x69, 0x6d, 0x65, 0x22, + 0x33, 0x0a, 0x10, 0x4e, 0x65, 0x67, 0x6f, 0x74, 0x69, 0x61, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, + 0x65, 0x73, 0x74, 0x12, 0x1f, 0x0a, 0x0b, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x5f, 0x68, 0x6f, 0x73, + 0x74, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0a, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x48, + 0x6f, 0x73, 0x74, 0x73, 0x22, 0x26, 0x0a, 0x0e, 0x4e, 0x65, 0x67, 0x6f, 0x74, 0x69, 0x61, 0x74, + 0x65, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x68, 0x6f, 0x73, 0x74, 0x73, 0x18, + 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x05, 0x68, 0x6f, 0x73, 0x74, 0x73, 0x22, 0x21, 0x0a, 0x0d, + 0x41, 0x63, 0x74, 0x6f, 0x72, 0x49, 0x64, 0x73, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x10, 0x0a, + 0x03, 0x69, 0x64, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x04, 0x52, 0x03, 0x69, 0x64, 0x73, 0x22, + 0x46, 0x0a, 0x0e, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x41, 0x63, + 0x6b, 0x12, 0x1a, 0x0a, 0x08, 0x61, 0x63, 0x63, 0x65, 0x70, 0x74, 0x65, 0x64, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x08, 0x52, 0x08, 0x61, 0x63, 0x63, 0x65, 0x70, 0x74, 0x65, 0x64, 0x12, 0x18, 0x0a, + 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, + 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x23, 0x0a, 0x0d, 0x43, 0x6c, 0x6f, 0x73, 0x69, + 0x6e, 0x67, 0x4e, 0x6f, 0x74, 0x69, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, + 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x22, 0x5c, 0x0a, 0x11, + 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x73, 0x68, 0x69, 0x70, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, + 0x65, 0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x04, 0x68, 0x6f, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x69, 0x64, 0x73, 0x18, 0x02, 0x20, 0x03, + 0x28, 0x04, 0x52, 0x03, 0x69, 0x64, 0x73, 0x12, 0x21, 0x0a, 0x0c, 0x6c, 0x61, 0x73, 0x74, 0x5f, + 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x03, 0x52, 0x0b, 0x6c, + 0x61, 0x73, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x73, 0x22, 0x59, 0x0a, 0x0e, 0x45, 0x78, + 0x70, 0x69, 0x72, 0x79, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, + 0x68, 0x6f, 0x73, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, + 0x12, 0x10, 0x0a, 0x03, 0x69, 0x64, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x04, 0x52, 0x03, 0x69, + 0x64, 0x73, 0x12, 0x21, 0x0a, 0x0c, 0x6c, 0x61, 0x73, 0x74, 0x5f, 0x63, 0x68, 0x61, 0x6e, 0x67, + 0x65, 0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x03, 0x52, 0x0b, 0x6c, 0x61, 0x73, 0x74, 0x43, 0x68, + 0x61, 0x6e, 0x67, 0x65, 0x73, 0x22, 0x50, 0x0a, 0x0c, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x52, 0x65, + 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x04, 0x52, 0x02, 0x69, 0x64, 0x12, 0x30, 0x0a, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, + 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, + 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x08, 0x6d, + 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x22, 0x1c, 0x0a, 0x0a, 0x47, 0x65, 0x74, 0x52, 0x65, + 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x04, 0x52, 0x02, 0x69, 0x64, 0x22, 0x36, 0x0a, 0x08, 0x47, 0x65, 0x74, 0x52, 0x65, 0x70, 0x6c, + 0x79, 0x12, 0x2a, 0x0a, 0x05, 0x67, 0x72, 0x61, 0x69, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, + 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, + 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x05, 0x67, 0x72, 0x61, 0x69, 0x6e, 0x22, 0x79, 0x0a, + 0x0e, 0x4d, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, + 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, + 0x79, 0x70, 0x65, 0x12, 0x2e, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, + 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, + 0x61, 0x67, 0x65, 0x12, 0x19, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x03, 0x20, 0x01, + 0x28, 0x09, 0x48, 0x00, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x88, 0x01, 0x01, 0x42, 0x08, + 0x0a, 0x06, 0x5f, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x22, 0x7f, 0x0a, 0x0b, 0x41, 0x70, 0x70, 0x6c, + 0x79, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x2a, 0x0a, 0x05, 0x73, 0x74, 0x61, 0x74, 0x65, + 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x05, 0x73, 0x74, + 0x61, 0x74, 0x65, 0x12, 0x44, 0x0a, 0x09, 0x6d, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, + 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, + 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, + 0x4d, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x09, + 0x6d, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x32, 0xd6, 0x05, 0x0a, 0x0c, 0x43, 0x6f, + 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x50, 0x6c, 0x61, 0x6e, 0x65, 0x12, 0x48, 0x0a, 0x04, 0x50, 0x69, + 0x6e, 0x67, 0x12, 0x1d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, + 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x45, 0x6d, 0x70, 0x74, + 0x79, 0x1a, 0x21, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, + 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, + 0x65, 0x70, 0x6c, 0x79, 0x12, 0x5d, 0x0a, 0x09, 0x4e, 0x65, 0x67, 0x6f, 0x74, 0x69, 0x61, 0x74, + 0x65, 0x12, 0x28, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, + 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4e, 0x65, 0x67, 0x6f, 0x74, + 0x69, 0x61, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x26, 0x2e, 0x63, 0x6f, + 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, + 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4e, 0x65, 0x67, 0x6f, 0x74, 0x69, 0x61, 0x74, 0x65, 0x52, 0x65, + 0x70, 0x6c, 0x79, 0x12, 0x58, 0x0a, 0x10, 0x47, 0x65, 0x74, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x41, + 0x63, 0x74, 0x6f, 0x72, 0x49, 0x64, 0x73, 0x12, 0x1d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, + 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, + 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x25, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, + 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, + 0x41, 0x63, 0x74, 0x6f, 0x72, 0x49, 0x64, 0x73, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x66, 0x0a, + 0x11, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x73, 0x68, + 0x69, 0x70, 0x12, 0x29, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, + 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4f, 0x77, 0x6e, 0x65, + 0x72, 0x73, 0x68, 0x69, 0x70, 0x41, 0x6e, 0x6e, 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x1a, 0x26, 0x2e, + 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, + 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x43, 0x68, 0x61, 0x6e, + 0x67, 0x65, 0x41, 0x63, 0x6b, 0x12, 0x52, 0x0a, 0x05, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x12, 0x24, + 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, + 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x41, 0x70, 0x70, 0x6c, 0x79, 0x52, 0x65, 0x71, + 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, + 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x41, 0x70, + 0x70, 0x6c, 0x79, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x60, 0x0a, 0x0e, 0x41, 0x6e, 0x6e, + 0x6f, 0x75, 0x6e, 0x63, 0x65, 0x45, 0x78, 0x70, 0x69, 0x72, 0x79, 0x12, 0x26, 0x2e, 0x63, 0x6f, + 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, + 0x61, 0x67, 0x65, 0x73, 0x2e, 0x45, 0x78, 0x70, 0x69, 0x72, 0x79, 0x41, 0x6e, 0x6e, 0x6f, 0x75, + 0x6e, 0x63, 0x65, 0x1a, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, + 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4f, 0x77, 0x6e, + 0x65, 0x72, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x41, 0x63, 0x6b, 0x12, 0x58, 0x0a, 0x07, 0x43, + 0x6c, 0x6f, 0x73, 0x69, 0x6e, 0x67, 0x12, 0x25, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, + 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, + 0x43, 0x6c, 0x6f, 0x73, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x74, 0x69, 0x63, 0x65, 0x1a, 0x26, 0x2e, + 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, + 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x43, 0x68, 0x61, 0x6e, + 0x67, 0x65, 0x41, 0x63, 0x6b, 0x12, 0x4b, 0x0a, 0x03, 0x47, 0x65, 0x74, 0x12, 0x22, 0x2e, 0x63, + 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, 0x5f, 0x6d, 0x65, 0x73, + 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, + 0x1a, 0x20, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, + 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x47, 0x65, 0x74, 0x52, 0x65, 0x70, + 0x6c, 0x79, 0x42, 0x45, 0x5a, 0x43, 0x67, 0x69, 0x74, 0x2e, 0x6b, 0x36, 0x6e, 0x2e, 0x6e, 0x65, + 0x74, 0x2f, 0x6d, 0x61, 0x74, 0x73, 0x2f, 0x67, 0x6f, 0x2d, 0x63, 0x61, 0x72, 0x74, 0x2d, 0x61, + 0x63, 0x74, 0x6f, 0x72, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, + 0x6f, 0x6c, 0x3b, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x5f, 0x70, 0x6c, 0x61, 0x6e, 0x65, + 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, + 0x33, +}) + +var ( + file_control_plane_proto_rawDescOnce sync.Once + file_control_plane_proto_rawDescData []byte +) + +func file_control_plane_proto_rawDescGZIP() []byte { + file_control_plane_proto_rawDescOnce.Do(func() { + file_control_plane_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_control_plane_proto_rawDesc), len(file_control_plane_proto_rawDesc))) + }) + return file_control_plane_proto_rawDescData +} + +var file_control_plane_proto_msgTypes = make([]protoimpl.MessageInfo, 14) +var file_control_plane_proto_goTypes = []any{ + (*Empty)(nil), // 0: control_plane_messages.Empty + (*PingReply)(nil), // 1: control_plane_messages.PingReply + (*NegotiateRequest)(nil), // 2: control_plane_messages.NegotiateRequest + (*NegotiateReply)(nil), // 3: control_plane_messages.NegotiateReply + (*ActorIdsReply)(nil), // 4: control_plane_messages.ActorIdsReply + (*OwnerChangeAck)(nil), // 5: control_plane_messages.OwnerChangeAck + (*ClosingNotice)(nil), // 6: control_plane_messages.ClosingNotice + (*OwnershipAnnounce)(nil), // 7: control_plane_messages.OwnershipAnnounce + (*ExpiryAnnounce)(nil), // 8: control_plane_messages.ExpiryAnnounce + (*ApplyRequest)(nil), // 9: control_plane_messages.ApplyRequest + (*GetRequest)(nil), // 10: control_plane_messages.GetRequest + (*GetReply)(nil), // 11: control_plane_messages.GetReply + (*MutationResult)(nil), // 12: control_plane_messages.MutationResult + (*ApplyResult)(nil), // 13: control_plane_messages.ApplyResult + (*anypb.Any)(nil), // 14: google.protobuf.Any +} +var file_control_plane_proto_depIdxs = []int32{ + 14, // 0: control_plane_messages.ApplyRequest.messages:type_name -> google.protobuf.Any + 14, // 1: control_plane_messages.GetReply.grain:type_name -> google.protobuf.Any + 14, // 2: control_plane_messages.MutationResult.message:type_name -> google.protobuf.Any + 14, // 3: control_plane_messages.ApplyResult.state:type_name -> google.protobuf.Any + 12, // 4: control_plane_messages.ApplyResult.mutations:type_name -> control_plane_messages.MutationResult + 0, // 5: control_plane_messages.ControlPlane.Ping:input_type -> control_plane_messages.Empty + 2, // 6: control_plane_messages.ControlPlane.Negotiate:input_type -> control_plane_messages.NegotiateRequest + 0, // 7: control_plane_messages.ControlPlane.GetLocalActorIds:input_type -> control_plane_messages.Empty + 7, // 8: control_plane_messages.ControlPlane.AnnounceOwnership:input_type -> control_plane_messages.OwnershipAnnounce + 9, // 9: control_plane_messages.ControlPlane.Apply:input_type -> control_plane_messages.ApplyRequest + 8, // 10: control_plane_messages.ControlPlane.AnnounceExpiry:input_type -> control_plane_messages.ExpiryAnnounce + 6, // 11: control_plane_messages.ControlPlane.Closing:input_type -> control_plane_messages.ClosingNotice + 10, // 12: control_plane_messages.ControlPlane.Get:input_type -> control_plane_messages.GetRequest + 1, // 13: control_plane_messages.ControlPlane.Ping:output_type -> control_plane_messages.PingReply + 3, // 14: control_plane_messages.ControlPlane.Negotiate:output_type -> control_plane_messages.NegotiateReply + 4, // 15: control_plane_messages.ControlPlane.GetLocalActorIds:output_type -> control_plane_messages.ActorIdsReply + 5, // 16: control_plane_messages.ControlPlane.AnnounceOwnership:output_type -> control_plane_messages.OwnerChangeAck + 13, // 17: control_plane_messages.ControlPlane.Apply:output_type -> control_plane_messages.ApplyResult + 5, // 18: control_plane_messages.ControlPlane.AnnounceExpiry:output_type -> control_plane_messages.OwnerChangeAck + 5, // 19: control_plane_messages.ControlPlane.Closing:output_type -> control_plane_messages.OwnerChangeAck + 11, // 20: control_plane_messages.ControlPlane.Get:output_type -> control_plane_messages.GetReply + 13, // [13:21] is the sub-list for method output_type + 5, // [5:13] is the sub-list for method input_type + 5, // [5:5] is the sub-list for extension type_name + 5, // [5:5] is the sub-list for extension extendee + 0, // [0:5] is the sub-list for field type_name +} + +func init() { file_control_plane_proto_init() } +func file_control_plane_proto_init() { + if File_control_plane_proto != nil { + return + } + file_control_plane_proto_msgTypes[12].OneofWrappers = []any{} + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_control_plane_proto_rawDesc), len(file_control_plane_proto_rawDesc)), + NumEnums: 0, + NumMessages: 14, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_control_plane_proto_goTypes, + DependencyIndexes: file_control_plane_proto_depIdxs, + MessageInfos: file_control_plane_proto_msgTypes, + }.Build() + File_control_plane_proto = out.File + file_control_plane_proto_goTypes = nil + file_control_plane_proto_depIdxs = nil +} diff --git a/proto/control_plane.proto b/proto/control_plane.proto index 31af1e1..c565b5b 100644 --- a/proto/control_plane.proto +++ b/proto/control_plane.proto @@ -55,16 +55,28 @@ message ClosingNotice { } // OwnershipAnnounce broadcasts first-touch ownership claims for cart IDs. -// First claim wins; receivers SHOULD NOT overwrite an existing different owner. +// Receivers arbitrate concurrent cold-cache first-touch claims using +// last_changes: the pod whose local grain has the earliest lastChange +// wins; ties resolve to the lexicographically smaller hostname. message OwnershipAnnounce { - string host = 1; // announcing host - repeated uint64 ids = 2; // newly claimed cart ids + string host = 1; // announcing host + repeated uint64 ids = 2; // newly claimed cart ids + // last_changes is the per-id UnixNano timestamp of the announcing + // pod's local grain's lastChange at the moment of broadcast. -1 + // means the announcer didn't have a local grain for this id (legacy + // path / TakeOwnership without prior load); receivers treat -1 as + // "no arbitration done" and accept the remote claim as before. The + // slice is aligned 1:1 with ids. + repeated int64 last_changes = 3; } // ExpiryAnnounce broadcasts that a host evicted the provided cart IDs. +// last_changes is carried for wire-shape symmetry with OwnershipAnnounce +// and is informational here (expiry is unilateral). message ExpiryAnnounce { string host = 1; repeated uint64 ids = 2; + repeated int64 last_changes = 3; } message ApplyRequest { diff --git a/proto/control_plane_grpc.pb.go b/proto/control_plane_grpc.pb.go new file mode 100644 index 0000000..617086b --- /dev/null +++ b/proto/control_plane_grpc.pb.go @@ -0,0 +1,403 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.5.1 +// - protoc v7.35.1 +// source: control_plane.proto + +package control_plane_messages + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + ControlPlane_Ping_FullMethodName = "/control_plane_messages.ControlPlane/Ping" + ControlPlane_Negotiate_FullMethodName = "/control_plane_messages.ControlPlane/Negotiate" + ControlPlane_GetLocalActorIds_FullMethodName = "/control_plane_messages.ControlPlane/GetLocalActorIds" + ControlPlane_AnnounceOwnership_FullMethodName = "/control_plane_messages.ControlPlane/AnnounceOwnership" + ControlPlane_Apply_FullMethodName = "/control_plane_messages.ControlPlane/Apply" + ControlPlane_AnnounceExpiry_FullMethodName = "/control_plane_messages.ControlPlane/AnnounceExpiry" + ControlPlane_Closing_FullMethodName = "/control_plane_messages.ControlPlane/Closing" + ControlPlane_Get_FullMethodName = "/control_plane_messages.ControlPlane/Get" +) + +// ControlPlaneClient is the client API for ControlPlane service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +// +// ControlPlane defines cluster coordination and ownership operations. +type ControlPlaneClient interface { + // Ping for liveness; lightweight health signal. + Ping(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*PingReply, error) + // Negotiate merges host views; used during discovery & convergence. + Negotiate(ctx context.Context, in *NegotiateRequest, opts ...grpc.CallOption) (*NegotiateReply, error) + // GetCartIds lists currently owned cart IDs on this node. + GetLocalActorIds(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*ActorIdsReply, error) + // Ownership announcement: first-touch claim broadcast (idempotent; best-effort). + AnnounceOwnership(ctx context.Context, in *OwnershipAnnounce, opts ...grpc.CallOption) (*OwnerChangeAck, error) + Apply(ctx context.Context, in *ApplyRequest, opts ...grpc.CallOption) (*ApplyResult, error) + // Expiry announcement: drop remote ownership hints when local TTL expires. + AnnounceExpiry(ctx context.Context, in *ExpiryAnnounce, opts ...grpc.CallOption) (*OwnerChangeAck, error) + // Closing announces graceful shutdown so peers can proactively adjust. + Closing(ctx context.Context, in *ClosingNotice, opts ...grpc.CallOption) (*OwnerChangeAck, error) + Get(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (*GetReply, error) +} + +type controlPlaneClient struct { + cc grpc.ClientConnInterface +} + +func NewControlPlaneClient(cc grpc.ClientConnInterface) ControlPlaneClient { + return &controlPlaneClient{cc} +} + +func (c *controlPlaneClient) Ping(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*PingReply, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(PingReply) + err := c.cc.Invoke(ctx, ControlPlane_Ping_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *controlPlaneClient) Negotiate(ctx context.Context, in *NegotiateRequest, opts ...grpc.CallOption) (*NegotiateReply, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(NegotiateReply) + err := c.cc.Invoke(ctx, ControlPlane_Negotiate_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *controlPlaneClient) GetLocalActorIds(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*ActorIdsReply, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(ActorIdsReply) + err := c.cc.Invoke(ctx, ControlPlane_GetLocalActorIds_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *controlPlaneClient) AnnounceOwnership(ctx context.Context, in *OwnershipAnnounce, opts ...grpc.CallOption) (*OwnerChangeAck, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(OwnerChangeAck) + err := c.cc.Invoke(ctx, ControlPlane_AnnounceOwnership_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *controlPlaneClient) Apply(ctx context.Context, in *ApplyRequest, opts ...grpc.CallOption) (*ApplyResult, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(ApplyResult) + err := c.cc.Invoke(ctx, ControlPlane_Apply_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *controlPlaneClient) AnnounceExpiry(ctx context.Context, in *ExpiryAnnounce, opts ...grpc.CallOption) (*OwnerChangeAck, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(OwnerChangeAck) + err := c.cc.Invoke(ctx, ControlPlane_AnnounceExpiry_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *controlPlaneClient) Closing(ctx context.Context, in *ClosingNotice, opts ...grpc.CallOption) (*OwnerChangeAck, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(OwnerChangeAck) + err := c.cc.Invoke(ctx, ControlPlane_Closing_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *controlPlaneClient) Get(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (*GetReply, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(GetReply) + err := c.cc.Invoke(ctx, ControlPlane_Get_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +// ControlPlaneServer is the server API for ControlPlane service. +// All implementations must embed UnimplementedControlPlaneServer +// for forward compatibility. +// +// ControlPlane defines cluster coordination and ownership operations. +type ControlPlaneServer interface { + // Ping for liveness; lightweight health signal. + Ping(context.Context, *Empty) (*PingReply, error) + // Negotiate merges host views; used during discovery & convergence. + Negotiate(context.Context, *NegotiateRequest) (*NegotiateReply, error) + // GetCartIds lists currently owned cart IDs on this node. + GetLocalActorIds(context.Context, *Empty) (*ActorIdsReply, error) + // Ownership announcement: first-touch claim broadcast (idempotent; best-effort). + AnnounceOwnership(context.Context, *OwnershipAnnounce) (*OwnerChangeAck, error) + Apply(context.Context, *ApplyRequest) (*ApplyResult, error) + // Expiry announcement: drop remote ownership hints when local TTL expires. + AnnounceExpiry(context.Context, *ExpiryAnnounce) (*OwnerChangeAck, error) + // Closing announces graceful shutdown so peers can proactively adjust. + Closing(context.Context, *ClosingNotice) (*OwnerChangeAck, error) + Get(context.Context, *GetRequest) (*GetReply, error) + mustEmbedUnimplementedControlPlaneServer() +} + +// UnimplementedControlPlaneServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedControlPlaneServer struct{} + +func (UnimplementedControlPlaneServer) Ping(context.Context, *Empty) (*PingReply, error) { + return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented") +} +func (UnimplementedControlPlaneServer) Negotiate(context.Context, *NegotiateRequest) (*NegotiateReply, error) { + return nil, status.Errorf(codes.Unimplemented, "method Negotiate not implemented") +} +func (UnimplementedControlPlaneServer) GetLocalActorIds(context.Context, *Empty) (*ActorIdsReply, error) { + return nil, status.Errorf(codes.Unimplemented, "method GetLocalActorIds not implemented") +} +func (UnimplementedControlPlaneServer) AnnounceOwnership(context.Context, *OwnershipAnnounce) (*OwnerChangeAck, error) { + return nil, status.Errorf(codes.Unimplemented, "method AnnounceOwnership not implemented") +} +func (UnimplementedControlPlaneServer) Apply(context.Context, *ApplyRequest) (*ApplyResult, error) { + return nil, status.Errorf(codes.Unimplemented, "method Apply not implemented") +} +func (UnimplementedControlPlaneServer) AnnounceExpiry(context.Context, *ExpiryAnnounce) (*OwnerChangeAck, error) { + return nil, status.Errorf(codes.Unimplemented, "method AnnounceExpiry not implemented") +} +func (UnimplementedControlPlaneServer) Closing(context.Context, *ClosingNotice) (*OwnerChangeAck, error) { + return nil, status.Errorf(codes.Unimplemented, "method Closing not implemented") +} +func (UnimplementedControlPlaneServer) Get(context.Context, *GetRequest) (*GetReply, error) { + return nil, status.Errorf(codes.Unimplemented, "method Get not implemented") +} +func (UnimplementedControlPlaneServer) mustEmbedUnimplementedControlPlaneServer() {} +func (UnimplementedControlPlaneServer) testEmbeddedByValue() {} + +// UnsafeControlPlaneServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to ControlPlaneServer will +// result in compilation errors. +type UnsafeControlPlaneServer interface { + mustEmbedUnimplementedControlPlaneServer() +} + +func RegisterControlPlaneServer(s grpc.ServiceRegistrar, srv ControlPlaneServer) { + // If the following call pancis, it indicates UnimplementedControlPlaneServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&ControlPlane_ServiceDesc, srv) +} + +func _ControlPlane_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(Empty) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ControlPlaneServer).Ping(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ControlPlane_Ping_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ControlPlaneServer).Ping(ctx, req.(*Empty)) + } + return interceptor(ctx, in, info, handler) +} + +func _ControlPlane_Negotiate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(NegotiateRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ControlPlaneServer).Negotiate(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ControlPlane_Negotiate_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ControlPlaneServer).Negotiate(ctx, req.(*NegotiateRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _ControlPlane_GetLocalActorIds_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(Empty) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ControlPlaneServer).GetLocalActorIds(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ControlPlane_GetLocalActorIds_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ControlPlaneServer).GetLocalActorIds(ctx, req.(*Empty)) + } + return interceptor(ctx, in, info, handler) +} + +func _ControlPlane_AnnounceOwnership_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(OwnershipAnnounce) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ControlPlaneServer).AnnounceOwnership(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ControlPlane_AnnounceOwnership_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ControlPlaneServer).AnnounceOwnership(ctx, req.(*OwnershipAnnounce)) + } + return interceptor(ctx, in, info, handler) +} + +func _ControlPlane_Apply_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ApplyRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ControlPlaneServer).Apply(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ControlPlane_Apply_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ControlPlaneServer).Apply(ctx, req.(*ApplyRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _ControlPlane_AnnounceExpiry_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ExpiryAnnounce) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ControlPlaneServer).AnnounceExpiry(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ControlPlane_AnnounceExpiry_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ControlPlaneServer).AnnounceExpiry(ctx, req.(*ExpiryAnnounce)) + } + return interceptor(ctx, in, info, handler) +} + +func _ControlPlane_Closing_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ClosingNotice) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ControlPlaneServer).Closing(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ControlPlane_Closing_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ControlPlaneServer).Closing(ctx, req.(*ClosingNotice)) + } + return interceptor(ctx, in, info, handler) +} + +func _ControlPlane_Get_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(GetRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ControlPlaneServer).Get(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ControlPlane_Get_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ControlPlaneServer).Get(ctx, req.(*GetRequest)) + } + return interceptor(ctx, in, info, handler) +} + +// ControlPlane_ServiceDesc is the grpc.ServiceDesc for ControlPlane service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var ControlPlane_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "control_plane_messages.ControlPlane", + HandlerType: (*ControlPlaneServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "Ping", + Handler: _ControlPlane_Ping_Handler, + }, + { + MethodName: "Negotiate", + Handler: _ControlPlane_Negotiate_Handler, + }, + { + MethodName: "GetLocalActorIds", + Handler: _ControlPlane_GetLocalActorIds_Handler, + }, + { + MethodName: "AnnounceOwnership", + Handler: _ControlPlane_AnnounceOwnership_Handler, + }, + { + MethodName: "Apply", + Handler: _ControlPlane_Apply_Handler, + }, + { + MethodName: "AnnounceExpiry", + Handler: _ControlPlane_AnnounceExpiry_Handler, + }, + { + MethodName: "Closing", + Handler: _ControlPlane_Closing_Handler, + }, + { + MethodName: "Get", + Handler: _ControlPlane_Get_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "control_plane.proto", +}