test

Makefile

@@ -19,7 +19,7 @@
 
 MODULE_PATH := git.tornberg.me/go-cart-actor
 PROTO_DIR   := proto
-PROTOS      := $(PROTO_DIR)/messages.proto $(PROTO_DIR)/cart_actor.proto $(PROTO_DIR)/control_plane.proto
+PROTOS      := $(PROTO_DIR)/messages.proto $(PROTO_DIR)/control_plane.proto
 
 # Allow override: make PROTOC=/path/to/protoc
 PROTOC ?= protoc
@@ -69,8 +69,8 @@ check_tools:
 protogen: check_tools
 	@echo "$(YELLOW)Generating protobuf code (outputs -> ./proto)...$(RESET)"
 	$(PROTOC) -I $(PROTO_DIR) \
-	  --go_out=./proto --go_opt=paths=source_relative \
-	  --go-grpc_out=./proto --go-grpc_opt=paths=source_relative \
+	  --go_out=./pkg/messages --go_opt=paths=source_relative \
+	  --go-grpc_out=./pkg/messages --go-grpc_opt=paths=source_relative \
 	  $(PROTOS)
 	@echo "$(GREEN)Protobuf generation complete.$(RESET)"
 

cart-grain-pool.go

@@ -0,0 +1,540 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"maps"
+	"reflect"
+	"sync"
+	"time"
+
+	"git.tornberg.me/go-cart-actor/pkg/actor"
+	"git.tornberg.me/go-cart-actor/pkg/discovery"
+	"git.tornberg.me/go-cart-actor/pkg/proxy"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"k8s.io/apimachinery/pkg/watch"
+)
+
+// ---------------------------------------------------------------------------
+// Metrics shared by the cart pool implementation.
+// ---------------------------------------------------------------------------
+
+var (
+	poolGrains = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "cart_grains_in_pool",
+		Help: "The total number of grains in the local pool",
+	})
+	poolSize = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "cart_pool_size",
+		Help: "Configured capacity of the cart pool",
+	})
+	poolUsage = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "cart_grain_pool_usage",
+		Help: "Current utilisation of the cart pool",
+	})
+	negotiationCount = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cart_remote_negotiation_total",
+		Help: "The total number of remote host negotiations",
+	})
+	connectedRemotes = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "cart_connected_remotes",
+		Help: "Number of connected remote hosts",
+	})
+	cartMutationsTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cart_mutations_total",
+		Help: "Total number of cart state mutations applied",
+	})
+	cartMutationFailuresTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cart_mutation_failures_total",
+		Help: "Total number of failed cart state mutations",
+	})
+	cartMutationLatencySeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "cart_mutation_latency_seconds",
+		Help:    "Latency of cart mutations in seconds",
+		Buckets: prometheus.DefBuckets,
+	}, []string{"mutation"})
+)
+
+// GrainPool is the interface exposed to HTTP handlers and other subsystems.
+
+// CartPool merges the responsibilities that previously belonged to
+// GrainLocalPool and SyncedPool. It provides local grain storage together
+// with cluster coordination, ownership negotiation and expiry signalling.
+type CartPool struct {
+	// Local grain state -----------------------------------------------------
+	localMu sync.RWMutex
+	grains  map[uint64]*CartGrain
+
+	spawn    func(id CartId) (*CartGrain, error)
+	ttl      time.Duration
+	poolSize int
+
+	// Cluster coordination --------------------------------------------------
+	hostname     string
+	remoteMu     sync.RWMutex
+	remoteOwners map[uint64]*proxy.RemoteHost
+	remoteHosts  map[string]*proxy.RemoteHost
+	//discardedHostHandler *DiscardedHostHandler
+
+	// House-keeping ---------------------------------------------------------
+	purgeTicker *time.Ticker
+}
+
+// NewCartPool constructs a unified pool. Discovery may be nil for standalone
+// deployments.
+func NewCartPool(size int, ttl time.Duration, hostname string, spawn func(id CartId) (*CartGrain, error), hostWatch discovery.Discovery) (*CartPool, error) {
+	p := &CartPool{
+		grains: make(map[uint64]*CartGrain),
+
+		spawn:        spawn,
+		ttl:          ttl,
+		poolSize:     size,
+		hostname:     hostname,
+		remoteOwners: make(map[uint64]*proxy.RemoteHost),
+		remoteHosts:  make(map[string]*proxy.RemoteHost),
+	}
+
+	p.purgeTicker = time.NewTicker(time.Minute)
+	go func() {
+		for range p.purgeTicker.C {
+			p.purge()
+		}
+	}()
+
+	if hostWatch != nil {
+		go p.startDiscovery(hostWatch)
+	} else {
+		log.Printf("No discovery configured; expecting manual AddRemote or static host injection")
+	}
+
+	return p, nil
+}
+
+func (p *CartPool) purge() {
+	purgeLimit := time.Now().Add(-p.ttl)
+	purgedIds := make([]uint64, 0, len(p.grains))
+	p.localMu.Lock()
+	for id, grain := range p.grains {
+		if grain.GetLastAccess().Before(purgeLimit) {
+			purgedIds = append(purgedIds, id)
+
+			delete(p.grains, id)
+		}
+	}
+	p.localMu.Unlock()
+	p.forAllHosts(func(remote *proxy.RemoteHost) {
+		remote.AnnounceExpiry(purgedIds)
+	})
+
+}
+
+// startDiscovery subscribes to cluster events and adds/removes hosts.
+func (p *CartPool) startDiscovery(discovery discovery.Discovery) {
+	time.Sleep(3 * time.Second) // allow gRPC server startup
+	log.Printf("Starting discovery watcher")
+	ch, err := discovery.Watch()
+	if err != nil {
+		log.Printf("Discovery error: %v", err)
+		return
+	}
+	for evt := range ch {
+		if evt.Host == "" {
+			continue
+		}
+		switch evt.Type {
+		case watch.Deleted:
+			if p.IsKnown(evt.Host) {
+				p.RemoveHost(evt.Host)
+			}
+		default:
+			if !p.IsKnown(evt.Host) {
+				log.Printf("Discovered host %s", evt.Host)
+				p.AddRemote(evt.Host)
+			}
+		}
+	}
+}
+
+// ---------------------------------------------------------------------------
+// Local grain management
+// ---------------------------------------------------------------------------
+
+func (p *CartPool) statsUpdate() {
+	p.localMu.RLock()
+	size := len(p.grains)
+	cap := p.poolSize
+	p.localMu.RUnlock()
+	poolGrains.Set(float64(size))
+	poolSize.Set(float64(cap))
+	if cap > 0 {
+		poolUsage.Set(float64(size) / float64(cap))
+	}
+}
+
+// LocalUsage returns the number of resident grains and configured capacity.
+func (p *CartPool) LocalUsage() (int, int) {
+	p.localMu.RLock()
+	defer p.localMu.RUnlock()
+	return len(p.grains), p.poolSize
+}
+
+// LocalCartIDs returns the currently owned cart ids (for control-plane RPCs).
+func (p *CartPool) GetLocalIds() []uint64 {
+	p.localMu.RLock()
+	defer p.localMu.RUnlock()
+	ids := make([]uint64, 0, len(p.grains))
+	for _, g := range p.grains {
+		if g == nil {
+			continue
+		}
+		ids = append(ids, uint64(g.GetId()))
+	}
+	return ids
+}
+
+func (p *CartPool) HandleRemoteExpiry(host string, ids []uint64) error {
+	p.remoteMu.Lock()
+	defer p.remoteMu.Unlock()
+	for _, id := range ids {
+		delete(p.remoteOwners, id)
+	}
+	return nil
+}
+
+func (p *CartPool) HandleOwnershipChange(host string, ids []uint64) error {
+
+	p.remoteMu.RLock()
+	remoteHost, exists := p.remoteHosts[host]
+	p.remoteMu.RUnlock()
+	if !exists {
+		createdHost, err := p.AddRemote(host)
+		if err != nil {
+			return err
+		}
+		remoteHost = createdHost
+	}
+	p.remoteMu.Lock()
+	defer p.remoteMu.Unlock()
+	p.localMu.Lock()
+	defer p.localMu.Unlock()
+	for _, id := range ids {
+		log.Printf("Handling ownership change for cart %d to host %s", id, host)
+		delete(p.grains, id)
+		p.remoteOwners[id] = remoteHost
+	}
+	return nil
+}
+
+// SnapshotGrains returns a copy of the currently resident grains keyed by id.
+func (p *CartPool) SnapshotGrains() map[uint64]*CartGrain {
+	p.localMu.RLock()
+	defer p.localMu.RUnlock()
+	out := maps.Clone(p.grains)
+
+	return out
+}
+
+// func (p *CartPool) getLocalGrain(key uint64) (*CartGrain, error) {
+
+// 	grainLookups.Inc()
+
+// 	p.localMu.RLock()
+// 	grain, ok := p.grains[key]
+// 	p.localMu.RUnlock()
+// 	if grain != nil && ok {
+// 		return grain, nil
+// 	}
+
+// 	go p.statsUpdate()
+// 	return grain, nil
+// }
+
+// ---------------------------------------------------------------------------
+// Cluster ownership and coordination
+// ---------------------------------------------------------------------------
+
+func (p *CartPool) TakeOwnership(id uint64) {
+
+	if p.grains[id] != nil {
+		return
+	}
+	log.Printf("taking ownership of: %d", id)
+	p.broadcastOwnership([]uint64{id})
+}
+
+func (p *CartPool) AddRemote(host string) (*proxy.RemoteHost, error) {
+	if host == "" || host == p.hostname || p.IsKnown(host) {
+		return nil, fmt.Errorf("invalid host")
+	}
+
+	remote, err := proxy.NewRemoteHost(host)
+	if err != nil {
+		log.Printf("AddRemote: NewRemoteHostGRPC %s failed: %v", host, err)
+
+		return nil, err
+	}
+	p.remoteMu.Lock()
+	p.remoteHosts[host] = remote
+	p.remoteMu.Unlock()
+	connectedRemotes.Set(float64(p.RemoteCount()))
+
+	log.Printf("Connected to remote host %s", host)
+	go p.pingLoop(remote)
+	go p.initializeRemote(remote)
+	go p.SendNegotiation()
+	return remote, nil
+}
+
+func (p *CartPool) initializeRemote(remote *proxy.RemoteHost) {
+
+	remotesIds := remote.GetActorIds()
+
+	p.remoteMu.Lock()
+	for _, id := range remotesIds {
+		p.localMu.Lock()
+		delete(p.grains, id)
+		p.localMu.Unlock()
+		if _, exists := p.remoteOwners[id]; !exists {
+			p.remoteOwners[id] = remote
+		}
+	}
+	p.remoteMu.Unlock()
+
+}
+
+func (p *CartPool) RemoveHost(host string) {
+	p.remoteMu.Lock()
+	remote, exists := p.remoteHosts[host]
+	if exists {
+		go remote.Close()
+		delete(p.remoteHosts, host)
+	}
+	for id, owner := range p.remoteOwners {
+		if owner.Host == host {
+			delete(p.remoteOwners, id)
+		}
+	}
+	p.remoteMu.Unlock()
+
+	if exists {
+		remote.Close()
+	}
+	connectedRemotes.Set(float64(p.RemoteCount()))
+}
+
+func (p *CartPool) RemoteCount() int {
+	p.remoteMu.RLock()
+	defer p.remoteMu.RUnlock()
+	return len(p.remoteHosts)
+}
+
+// RemoteHostNames returns a snapshot of connected remote host identifiers.
+func (p *CartPool) RemoteHostNames() []string {
+	p.remoteMu.RLock()
+	defer p.remoteMu.RUnlock()
+	hosts := make([]string, 0, len(p.remoteHosts))
+	for host := range p.remoteHosts {
+		hosts = append(hosts, host)
+	}
+	return hosts
+}
+
+func (p *CartPool) IsKnown(host string) bool {
+	if host == p.hostname {
+		return true
+	}
+	p.remoteMu.RLock()
+	defer p.remoteMu.RUnlock()
+	_, ok := p.remoteHosts[host]
+	return ok
+}
+
+func (p *CartPool) pingLoop(remote *proxy.RemoteHost) {
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+	for range ticker.C {
+		if !remote.Ping() {
+			if !remote.IsHealthy() {
+				log.Printf("Remote %s unhealthy, removing", remote.Host)
+				p.Close()
+				p.RemoveHost(remote.Host)
+				return
+			}
+			continue
+		}
+	}
+}
+
+func (p *CartPool) IsHealthy() bool {
+	p.remoteMu.RLock()
+	defer p.remoteMu.RUnlock()
+	for _, r := range p.remoteHosts {
+		if !r.IsHealthy() {
+			return false
+		}
+	}
+	return true
+}
+
+func (p *CartPool) Negotiate(otherHosts []string) {
+
+	for _, host := range otherHosts {
+		if host != p.hostname {
+			p.remoteMu.RLock()
+			_, ok := p.remoteHosts[host]
+			p.remoteMu.RUnlock()
+			if !ok {
+				go p.AddRemote(host)
+			}
+		}
+	}
+}
+
+func (p *CartPool) SendNegotiation() {
+	negotiationCount.Inc()
+
+	p.remoteMu.RLock()
+	hosts := make([]string, 0, len(p.remoteHosts)+1)
+	hosts = append(hosts, p.hostname)
+	remotes := make([]*proxy.RemoteHost, 0, len(p.remoteHosts))
+	for h, r := range p.remoteHosts {
+		hosts = append(hosts, h)
+		remotes = append(remotes, r)
+	}
+	p.remoteMu.RUnlock()
+
+	p.forAllHosts(func(remote *proxy.RemoteHost) {
+		knownByRemote, err := remote.Negotiate(hosts)
+
+		if err != nil {
+			log.Printf("Negotiate with %s failed: %v", remote.Host, err)
+			return
+		}
+		for _, h := range knownByRemote {
+			if !p.IsKnown(h) {
+				go p.AddRemote(h)
+			}
+		}
+	})
+}
+
+func (p *CartPool) forAllHosts(fn func(*proxy.RemoteHost)) {
+	p.remoteMu.RLock()
+	rh := maps.Clone(p.remoteHosts)
+	p.remoteMu.RUnlock()
+
+	wg := sync.WaitGroup{}
+	for _, host := range rh {
+
+		wg.Go(func() { fn(host) })
+
+	}
+
+	for name, host := range rh {
+		if !host.IsHealthy() {
+			host.Close()
+			p.remoteMu.Lock()
+			delete(p.remoteHosts, name)
+			p.remoteMu.Unlock()
+		}
+
+	}
+}
+
+func (p *CartPool) broadcastOwnership(ids []uint64) {
+	if len(ids) == 0 {
+		return
+	}
+
+	p.forAllHosts(func(rh *proxy.RemoteHost) {
+		rh.AnnounceOwnership(ids)
+	})
+
+}
+
+func (p *CartPool) getOrClaimGrain(id uint64) (*CartGrain, error) {
+	p.localMu.RLock()
+	grain, exists := p.grains[id]
+	p.localMu.RUnlock()
+	if exists && grain != nil {
+		return grain, nil
+	}
+
+	grain, err := p.spawn(CartId(id))
+	if err != nil {
+		return nil, err
+	}
+	go p.broadcastOwnership([]uint64{id})
+	return grain, nil
+}
+
+// ErrNotOwner is returned when a cart belongs to another host.
+var ErrNotOwner = fmt.Errorf("not owner")
+
+// Apply applies a mutation to a grain.
+func (p *CartPool) Apply(id uint64, mutation any) (*CartGrain, error) {
+	grain, err := p.getOrClaimGrain(id)
+	if err != nil {
+		return nil, err
+	}
+	start := time.Now()
+	result, applyErr := grain.Apply(mutation, false)
+	mutationType := "unknown"
+	if mutation != nil {
+		if t := reflect.TypeOf(mutation); t != nil {
+			if t.Kind() == reflect.Pointer {
+				t = t.Elem()
+			}
+			if t.Name() != "" {
+				mutationType = t.Name()
+			}
+		}
+	}
+	cartMutationLatencySeconds.WithLabelValues(mutationType).Observe(time.Since(start).Seconds())
+
+	if applyErr == nil && result != nil {
+		cartMutationsTotal.Inc()
+
+	} else if applyErr != nil {
+		cartMutationFailuresTotal.Inc()
+	}
+
+	return result, applyErr
+}
+
+// Get returns the current state of a grain.
+func (p *CartPool) Get(id uint64) (*CartGrain, error) {
+	grain, err := p.getOrClaimGrain(id)
+	if err != nil {
+		return nil, err
+	}
+	return grain.GetCurrentState()
+}
+
+// OwnerHost reports the remote owner (if any) for the supplied cart id.
+func (p *CartPool) OwnerHost(id uint64) (actor.Host, bool) {
+	p.remoteMu.RLock()
+	defer p.remoteMu.RUnlock()
+	owner, ok := p.remoteOwners[id]
+	return owner, ok
+}
+
+// Hostname returns the local hostname (pod IP).
+func (p *CartPool) Hostname() string {
+	return p.hostname
+}
+
+// Close notifies remotes that this host is shutting down.
+func (p *CartPool) Close() {
+	p.remoteMu.Lock()
+	defer p.remoteMu.Unlock()
+	for _, r := range p.remoteHosts {
+		go func(rh *proxy.RemoteHost) {
+			rh.Close()
+		}(r)
+	}
+	if p.purgeTicker != nil {
+		p.purgeTicker.Stop()
+	}
+}

cart-grain.go

@@ -3,42 +3,15 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"slices"
 	"sync"
+	"time"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
-type CartId [16]byte
-
-// String returns the cart id as a trimmed UTF-8 string (trailing zero bytes removed).
-func (id CartId) String() string {
-	n := 0
-	for n < len(id) && id[n] != 0 {
-		n++
-	}
-	return string(id[:n])
-}
-
-// ToCartId converts an arbitrary string to a fixed-size CartId (truncating or padding with zeros).
-func ToCartId(s string) CartId {
-	var id CartId
-	copy(id[:], []byte(s))
-	return id
-}
-
-func (id CartId) MarshalJSON() ([]byte, error) {
-	return json.Marshal(id.String())
-}
-
-func (id *CartId) UnmarshalJSON(data []byte) error {
-	var str string
-	err := json.Unmarshal(data, &str)
-	if err != nil {
-		return err
-	}
-	copy(id[:], []byte(str))
-	return nil
-}
+// Legacy padded [16]byte CartId and its helper methods removed.
+// Unified CartId (uint64 with base62 string form) now defined in cart_id.go.
 
 type StockStatus int
 
@@ -89,6 +62,8 @@ type CartGrain struct {
 	mu                sync.RWMutex
 	lastItemId        int
 	lastDeliveryId    int
+	lastAccess        time.Time
+	lastChange        time.Time       // unix seconds of last successful mutation (replay sets from event ts)
 	Id                CartId          `json:"id"`
 	Items             []*CartItem     `json:"items"`
 	TotalPrice        int64           `json:"totalPrice"`
@@ -101,22 +76,20 @@ type CartGrain struct {
 	PaymentStatus     string          `json:"paymentStatus,omitempty"`
 }
 
-type Grain interface {
-	GetId() CartId
-	Apply(content interface{}, isReplay bool) (*CartGrain, error)
-	GetCurrentState() (*CartGrain, error)
-}
-
 func (c *CartGrain) GetId() CartId {
 	return c.Id
 }
 
-func (c *CartGrain) GetLastChange() int64 {
-	// Legacy event log removed; return 0 to indicate no persisted mutation history.
-	return 0
+func (c *CartGrain) GetLastChange() time.Time {
+	return c.lastChange
+}
+
+func (c *CartGrain) GetLastAccess() time.Time {
+	return c.lastAccess
 }
 
 func (c *CartGrain) GetCurrentState() (*CartGrain, error) {
+	c.lastAccess = time.Now()
 	return c, nil
 }
 
@@ -141,12 +114,16 @@ func getItemData(sku string, qty int, country string) (*messages.AddItem, error)
 	}
 
 	stock := InStock
-	/*item.t
-	if item.StockLevel == "0" || item.StockLevel == "" {
+	item.HasStock()
+	stockValue, ok := item.GetNumberFieldValue(3)
+	if !ok || stockValue == 0 {
 		stock = OutOfStock
-	} else if item.StockLevel == "5+" {
+	} else {
+		if stockValue < 5 {
 			stock = LowStock
-		}*/
+		}
+	}
+
 	articleType, _ := item.GetStringFieldValue(1)   //.Fields[1].(string)
 	outletGrade, ok := item.GetStringFieldValue(20) //.Fields[20].(string)
 	var outlet *string
@@ -197,15 +174,6 @@ func (c *CartGrain) AddItem(sku string, qty int, country string, storeId *string
 	return c.Apply(cartItem, false)
 }
 
-/*
-Legacy storage (event sourcing) removed in oneof refactor.
-Kept stub (commented) for potential future reintroduction using proto envelopes.
-
-func (c *CartGrain) GetStorageMessage(since int64) []interface{} {
-	return nil
-}
-*/
-
 func (c *CartGrain) GetState() ([]byte, error) {
 	return json.Marshal(c)
 }
@@ -228,13 +196,7 @@ func (c *CartGrain) ItemsWithoutDelivery() []int {
 	ret := make([]int, 0, len(c.Items))
 	hasDelivery := c.ItemsWithDelivery()
 	for _, item := range c.Items {
-		found := false
-		for _, id := range hasDelivery {
-			if item.Id == id {
-				found = true
-				break
-			}
-		}
+		found := slices.Contains(hasDelivery, item.Id)
 
 		if !found {
 			ret = append(ret, item.Id)
@@ -269,6 +231,14 @@ func (c *CartGrain) Apply(content interface{}, isReplay bool) (*CartGrain, error
 		}
 		return nil, err
 	}
+
+	// Sliding TTL: update lastChange only for non-replay successful mutations.
+	if updated != nil && !isReplay {
+		c.lastChange = time.Now()
+		c.lastAccess = time.Now()
+		_ = AppendCartEvent(c.Id, content)
+	}
+
 	return updated, nil
 }
 

cart_id.go

@@ -2,63 +2,43 @@ package main
 
 import (
 	"crypto/rand"
-	"encoding/binary"
-	"errors"
+	"encoding/json"
 	"fmt"
-	"strings"
 )
 
 // cart_id.go
 //
-// Compact CartID implementation using 64 bits of cryptographic randomness,
-// base62 encoded (0-9 A-Z a-z). Typical length is 11 characters (since 62^11 > 2^64).
+// Breaking change:
+//   Unified cart identifier as a raw 64-bit unsigned integer (type CartId uint64).
+//   External textual representation: base62 (0-9 A-Z a-z), shortest possible
+//   encoding for 64 bits (max 11 characters, since 62^11 > 2^64).
 //
-// Motivation:
-//   * Shorter identifiers for cookies / URLs than legacy padded 16-byte CartId
-//   * O(1) hashing (raw uint64) for consistent hashing ring integration
-//   * Extremely low collision probability (birthday bound negligible at scale)
+// Rationale:
+//   - Replaces legacy fixed [16]byte padded string and transitional CartID wrapper.
+//   - Provides compact, URL/cookie-friendly identifiers.
+//   - O(1) hashing and minimal memory footprint.
+//   - 64 bits of crypto randomness => negligible collision probability at realistic scale.
 //
-// Backward Compatibility Strategy (Phased):
-//   Phase 1: Introduce CartID helpers while continuing to accept legacy CartId.
-//   Phase 2: Internally migrate maps to key by uint64 (CartID.Raw()).
-//   Phase 3: Canonicalize all inbound IDs to short base62; reissue Set-Cart-Id header.
+// Public API:
+//   type CartId uint64
+//   func NewCartId() (CartId, error)
+//   func MustNewCartId() CartId
+//   func ParseCartId(string) (CartId, bool)
+//   func MustParseCartId(string) CartId
+//   (CartId).String() string
+//   (CartId).MarshalJSON() / UnmarshalJSON()
 //
 // NOTE:
-//   The legacy type `CartId [16]byte` is still present elsewhere; helper
-//   UpgradeLegacyCartId bridges that representation to the new form without
-//   breaking deterministic mapping for existing carts.
-//
-// Security / Predictability:
-//   Uses crypto/rand for generation. If ever required, you can layer an
-//   HMAC-based derivation for additional secrecy. Current approach already
-//   provides 64 bits of entropy (brute force infeasible for practical risk).
-//
-// Future Extensions:
-//   * Time-sortable IDs: prepend a 48-bit timestamp field and encode 80 bits.
-//   * Add metrics counters for: generated_new, parsed_existing, legacy_fallback.
-//   * Add a pool of pre-generated IDs for ultra-low-latency hot paths (rarely needed).
-//
-// Public Surface Summary:
-//   NewCartID() (CartID, error)
-//   ParseCartID(string) (CartID, bool)
-//   FallbackFromString(string) CartID
-//   UpgradeLegacyCartId(CartId) CartID
-//   CanonicalizeIncoming(string) (CartID, bool /*wasGenerated*/, error)
-//
-// Encoding Details:
-//   encodeBase62 / decodeBase62 maintain a stable alphabet. DO NOT change
-//   alphabet order once IDs are in circulation, or previously issued IDs
-//   will change meaning.
-//
-// Zero Values:
-//   The zero value CartID{} has raw=0, txt="0". Treat it as valid but
-//   usually you will call NewCartID instead.
+//   All legacy helpers (UpgradeLegacyCartId, Fallback hashing, Canonicalize variants,
+//   CartIDToLegacy, LegacyToCartID) have been removed as part of the breaking change.
 //
 // ---------------------------------------------------------------------------
 
+type CartId uint64
+
 const base62Alphabet = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
 
-// Precomputed reverse lookup table for decode (255 = invalid).
+// Reverse lookup (0xFF marks invalid)
 var base62Rev [256]byte
 
 func init() {
@@ -70,136 +50,90 @@ func init() {
 	}
 }
 
-// CartID is the compact representation of a cart identifier.
-// raw: 64-bit entropy (also used directly for consistent hashing).
-// txt: cached base62 textual form.
-type CartID struct {
-	raw uint64
-	txt string
+// String returns the canonical base62 encoding of the 64-bit id.
+func (id CartId) String() string {
+	return encodeBase62(uint64(id))
 }
 
-// String returns the canonical base62 encoded ID.
-func (c CartID) String() string {
-	if c.txt == "" { // lazily encode if constructed manually
-		c.txt = encodeBase62(c.raw)
+// MarshalJSON encodes the cart id as a JSON string.
+func (id CartId) MarshalJSON() ([]byte, error) {
+	return json.Marshal(id.String())
+}
+
+// UnmarshalJSON decodes a cart id from a JSON string containing base62 text.
+func (id *CartId) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return err
 	}
-	return c.txt
+	parsed, ok := ParseCartId(s)
+	if !ok {
+		return fmt.Errorf("invalid cart id: %q", s)
+	}
+	*id = parsed
+	return nil
 }
 
-// Raw returns the 64-bit numeric value (useful for hashing / ring lookup).
-func (c CartID) Raw() uint64 {
-	return c.raw
-}
-
-// IsZero reports whether this CartID is the zero value.
-func (c CartID) IsZero() bool {
-	return c.raw == 0
-}
-
-// NewCartID generates a new cryptographically random 64-bit ID.
-func NewCartID() (CartID, error) {
+// NewCartId generates a new cryptographically random non-zero 64-bit id.
+func NewCartId() (CartId, error) {
 	var b [8]byte
 	if _, err := rand.Read(b[:]); err != nil {
-		return CartID{}, fmt.Errorf("NewCartID: %w", err)
+		return 0, fmt.Errorf("NewCartId: %w", err)
 	}
-	u := binary.BigEndian.Uint64(b[:])
-	// Reject zero if you want to avoid ever producing "0" (optional).
+	u := (uint64(b[0]) << 56) |
+		(uint64(b[1]) << 48) |
+		(uint64(b[2]) << 40) |
+		(uint64(b[3]) << 32) |
+		(uint64(b[4]) << 24) |
+		(uint64(b[5]) << 16) |
+		(uint64(b[6]) << 8) |
+		uint64(b[7])
 	if u == 0 {
-		// Extremely unlikely; recurse once.
-		return NewCartID()
+		// Extremely unlikely; regenerate once to avoid "0" identifier if desired.
+		return NewCartId()
 	}
-	return CartID{raw: u, txt: encodeBase62(u)}, nil
+	return CartId(u), nil
 }
 
-// MustNewCartID panics on failure (suitable for tests / initialization).
-func MustNewCartID() CartID {
-	id, err := NewCartID()
+// MustNewCartId panics if generation fails.
+func MustNewCartId() CartId {
+	id, err := NewCartId()
 	if err != nil {
 		panic(err)
 	}
 	return id
 }
 
-// ParseCartID attempts to parse a base62 canonical ID.
-// Returns (id, true) if fully valid; (zero, false) otherwise.
-func ParseCartID(s string) (CartID, bool) {
-	if len(s) == 0 {
-		return CartID{}, false
-	}
-	// Basic length sanity; allow a bit of headroom for future timestamp variant.
-	if len(s) > 16 {
-		return CartID{}, false
+// ParseCartId parses a base62 string into a CartId.
+// Returns (0,false) for invalid input.
+func ParseCartId(s string) (CartId, bool) {
+	// Accept length 1..11 (11 sufficient for 64 bits). Reject >11 immediately.
+	// Provide a slightly looser upper bound (<=16) only if you anticipate future
+	// extensions; here we stay strict.
+	if len(s) == 0 || len(s) > 11 {
+		return 0, false
 	}
 	u, ok := decodeBase62(s)
 	if !ok {
-		return CartID{}, false
+		return 0, false
 	}
-	return CartID{raw: u, txt: s}, true
+	return CartId(u), true
 }
 
-// FallbackFromString produces a deterministic CartID from arbitrary input
-// using a 64-bit FNV-1a hash. This allows legacy or malformed IDs to map
-// consistently into the new scheme (collision probability still low).
-func FallbackFromString(s string) CartID {
-	const (
-		offset64 = 1469598103934665603
-		prime64  = 1099511628211
-	)
-	h := uint64(offset64)
-	for i := 0; i < len(s); i++ {
-		h ^= uint64(s[i])
-		h *= prime64
+// MustParseCartId panics on invalid base62 input.
+func MustParseCartId(s string) CartId {
+	id, ok := ParseCartId(s)
+	if !ok {
+		panic(fmt.Sprintf("invalid cart id: %q", s))
 	}
-	return CartID{raw: h, txt: encodeBase62(h)}
+	return id
 }
 
-// UpgradeLegacyCartId converts the old 16-byte CartId (padded) to CartID
-// by hashing its trimmed string form. Keeps stable mapping across restarts.
-func UpgradeLegacyCartId(old CartId) CartID {
-	return FallbackFromString(old.String())
-}
-
-// CanonicalizeIncoming normalizes user-provided ID strings.
-// Behavior:
-//
-//	Empty string  -> generate new ID (wasGenerated = true)
-//	Valid base62  -> parse and return (wasGenerated = false)
-//	Anything else -> fallback deterministic hash (wasGenerated = false)
-//
-// Errors only occur if crypto/rand fails during generation.
-func CanonicalizeIncoming(s string) (CartID, bool, error) {
-	if s == "" {
-		id, err := NewCartID()
-		return id, true, err
-	}
-	if cid, ok := ParseCartID(s); ok {
-		return cid, false, nil
-	}
-	// Legacy heuristic: if length == 16 and contains non-base62 chars, treat as legacy padded ID.
-	if len(s) == 16 && !isAllBase62(s) {
-		return FallbackFromString(strings.TrimRight(s, "\x00")), false, nil
-	}
-	return FallbackFromString(s), false, nil
-}
-
-// isAllBase62 returns true if every byte is in the base62 alphabet.
-func isAllBase62(s string) bool {
-	for i := 0; i < len(s); i++ {
-		if base62Rev[s[i]] == 0xFF {
-			return false
-		}
-	}
-	return true
-}
-
-// encodeBase62 turns a uint64 into base62 text.
-// Complexity: O(log_62 n) ~ at most 11 iterations for 64 bits.
+// encodeBase62 converts a uint64 to base62 (shortest form).
 func encodeBase62(u uint64) string {
 	if u == 0 {
 		return "0"
 	}
-	// 62^11 = 743008370688 > 2^39; 62^11 > 2^64? Actually 62^11 ~= 5.18e19 < 2^64 (1.84e19)? 2^64 ≈ 1.84e19.
-	// 62^11 ≈ 5.18e19 > 2^64? Correction: 2^64 ≈ 1.844e19, so 62^11 > 2^64. Thus 11 chars suffice.
 	var buf [11]byte
 	i := len(buf)
 	for u > 0 {
@@ -210,8 +144,7 @@ func encodeBase62(u uint64) string {
 	return string(buf[i:])
 }
 
-// decodeBase62 converts a base62 string to uint64.
-// Returns (value, false) if any invalid character appears.
+// decodeBase62 converts base62 text to uint64.
 func decodeBase62(s string) (uint64, bool) {
 	var v uint64
 	for i := 0; i < len(s); i++ {
@@ -224,104 +157,3 @@ func decodeBase62(s string) (uint64, bool) {
 	}
 	return v, true
 }
-
-// ErrInvalidCartID can be returned by higher-level validation layers if you decide
-// to reject fallback-derived IDs (currently unused here).
-var ErrInvalidCartID = errors.New("invalid cart id")
-
-// ---------------------------------------------------------------------------
-// Legacy / Compatibility Conversion Helpers
-// ---------------------------------------------------------------------------
-
-// CartIDToLegacy converts a CartID (base62) into the legacy fixed-size CartId
-// ([16]byte) by copying the textual form (truncated or zero-padded).
-// NOTE: If the base62 string is longer than 16 (should not happen with current
-// 64-bit space), it will be truncated.
-func CartIDToLegacy(c CartID) CartId {
-	var id CartId
-	txt := c.String()
-	copy(id[:], []byte(txt))
-	return id
-}
-
-// LegacyToCartID upgrades a legacy CartId (padded) to a CartID by hashing its
-// trimmed string form (deterministic). This preserves stable mapping without
-// depending on original randomness.
-func LegacyToCartID(old CartId) CartID {
-	return UpgradeLegacyCartId(old)
-}
-
-// CartIDToKey returns the numeric key representation (uint64) for map indexing.
-func CartIDToKey(c CartID) uint64 {
-	return c.Raw()
-}
-
-// LegacyToCartKey converts a legacy CartId to the numeric key via deterministic
-// fallback hashing. (Uses the same logic as LegacyToCartID then returns raw.)
-func LegacyToCartKey(old CartId) uint64 {
-	return LegacyToCartID(old).Raw()
-}
-
-// ---------------------- Optional Helper Utilities ----------------------------
-
-// CartIDOrNew tries to parse s; if empty OR invalid returns a fresh ID.
-func CartIDOrNew(s string) (CartID, bool /*wasParsed*/, error) {
-	if cid, ok := ParseCartID(s); ok {
-		return cid, true, nil
-	}
-	id, err := NewCartID()
-	return id, false, err
-}
-
-// MustParseCartID panics if s is not a valid base62 ID (useful in tests).
-func MustParseCartID(s string) CartID {
-	if cid, ok := ParseCartID(s); ok {
-		return cid
-	}
-	panic(fmt.Sprintf("invalid CartID: %s", s))
-}
-
-// DebugString returns a verbose description (for logging / diagnostics).
-func (c CartID) DebugString() string {
-	return fmt.Sprintf("CartID(raw=%d txt=%s)", c.raw, c.String())
-}
-
-// Equal compares two CartIDs by raw value.
-func (c CartID) Equal(other CartID) bool {
-	return c.raw == other.raw
-}
-
-// CanonicalizeOrLegacy preserves legacy (non-base62) IDs without altering their
-// textual form, avoiding the previous behavior where fallback hashing replaced
-// the original string with a base62-encoded hash (which broke deterministic
-// key derivation across mixed call paths).
-//
-// Behavior:
-//   - s == ""      -> generate new CartID (generatedNew = true, wasBase62 = true)
-//   - base62 ok    -> return parsed CartID (generatedNew = false, wasBase62 = true)
-//   - otherwise    -> treat as legacy: raw = hash(s), txt = original s
-//
-// Returns:
-//
-//	cid          - CartID (txt preserved for legacy inputs)
-//	generatedNew - true only when a brand new ID was created due to empty input
-//	wasBase62    - true if the input was already canonical base62 (or generated)
-//	err          - only set if crypto/rand fails when generating a new ID
-func CanonicalizeOrLegacy(s string) (cid CartID, generatedNew bool, wasBase62 bool, err error) {
-	if s == "" {
-		id, e := NewCartID()
-		if e != nil {
-			return CartID{}, false, false, e
-		}
-		return id, true, true, nil
-	}
-	if parsed, ok := ParseCartID(s); ok {
-		return parsed, false, true, nil
-	}
-	// Legacy path: keep original text so downstream legacy-to-key hashing
-	// (which uses the visible string) yields consistent keys across code paths.
-	hashCID := FallbackFromString(s)
-	// Preserve original textual form
-	hashCID.txt = s
-	return hashCID, false, false, nil
-}

cart_id_test.go

@@ -1,168 +1,155 @@
 package main
 
 import (
-	"crypto/rand"
-	"encoding/binary"
+	"encoding/json"
 	"fmt"
-	mrand "math/rand"
 	"testing"
 )
 
-// TestEncodeDecodeBase62RoundTrip verifies encodeBase62/decodeBase62 are inverse.
-func TestEncodeDecodeBase62RoundTrip(t *testing.T) {
-	mrand.Seed(42)
-	for i := 0; i < 1000; i++ {
-		// Random 64-bit value
-		v := mrand.Uint64()
-		s := encodeBase62(v)
-		dec, ok := decodeBase62(s)
-		if !ok {
-			t.Fatalf("decodeBase62 failed for %d encoded=%s", v, s)
-		}
-		if dec != v {
-			t.Fatalf("round trip mismatch: have %d got %d (encoded=%s)", v, dec, s)
-		}
-	}
-	// Explicit zero test
-	if s := encodeBase62(0); s != "0" {
-		t.Fatalf("expected encodeBase62(0) == \"0\", got %q", s)
-	}
-	if v, ok := decodeBase62("0"); !ok || v != 0 {
-		t.Fatalf("decodeBase62(0) unexpected result v=%d ok=%v", v, ok)
-	}
-}
-
-// TestNewCartIDUniqueness generates a number of IDs and checks for duplicates.
-func TestNewCartIDUniqueness(t *testing.T) {
-	const n = 10000
+// TestNewCartIdUniqueness generates many ids and checks for collisions.
+func TestNewCartIdUniqueness(t *testing.T) {
+	const n = 20000
 	seen := make(map[string]struct{}, n)
 	for i := 0; i < n; i++ {
-		id, err := NewCartID()
+		id, err := NewCartId()
 		if err != nil {
-			t.Fatalf("NewCartID error: %v", err)
+			t.Fatalf("NewCartId error: %v", err)
 		}
 		s := id.String()
 		if _, exists := seen[s]; exists {
-			t.Fatalf("duplicate CartID generated: %s", s)
+			t.Fatalf("duplicate id encountered: %s", s)
 		}
 		seen[s] = struct{}{}
-		if id.IsZero() {
-			t.Fatalf("NewCartID returned zero value")
+		if s == "" {
+			t.Fatalf("empty string representation for id %d", id)
+		}
+		if len(s) > 11 {
+			t.Fatalf("encoded id length exceeds 11 chars: %s (%d)", s, len(s))
+		}
+		if id == 0 {
+			// We force regeneration on zero, extremely unlikely but test guards intent.
+			t.Fatalf("zero id generated (should be regenerated)")
 		}
 	}
 }
 
-// TestParseCartIDValidation tests parsing of valid and invalid base62 strings.
-func TestParseCartIDValidation(t *testing.T) {
-	id, err := NewCartID()
-	if err != nil {
-		t.Fatalf("NewCartID error: %v", err)
-	}
-	parsed, ok := ParseCartID(id.String())
+// TestParseCartIdRoundTrip ensures parse -> string -> parse is stable.
+func TestParseCartIdRoundTrip(t *testing.T) {
+	id := MustNewCartId()
+	txt := id.String()
+	parsed, ok := ParseCartId(txt)
 	if !ok {
-		t.Fatalf("ParseCartID failed for valid id %s", id)
+		t.Fatalf("ParseCartId failed for valid text %q", txt)
 	}
-	if parsed.raw != id.raw {
-		t.Fatalf("parsed raw mismatch: %d vs %d", parsed.raw, id.raw)
-	}
-
-	if _, ok := ParseCartID(""); ok {
-		t.Fatalf("expected empty string to be invalid")
-	}
-	// Invalid char ('-')
-	if _, ok := ParseCartID("abc-123"); ok {
-		t.Fatalf("expected invalid chars to fail parse")
-	}
-	// Overly long ( >16 )
-	if _, ok := ParseCartID("1234567890abcdefg"); ok {
-		t.Fatalf("expected overly long string to fail parse")
+	if parsed != id {
+		t.Fatalf("round trip mismatch: original=%d parsed=%d txt=%s", id, parsed, txt)
 	}
 }
 
-// TestFallbackDeterminism ensures fallback hashing is deterministic.
-func TestFallbackDeterminism(t *testing.T) {
-	inputs := []string{
-		"legacy-cart-1",
-		"legacy-cart-2",
-		"UPPER_lower_123",
-		"🚀unicode", // unicode bytes (will hash byte sequence)
+// TestParseCartIdInvalid covers invalid inputs.
+func TestParseCartIdInvalid(t *testing.T) {
+	invalid := []string{
+		"",               // empty
+		" ",              // space
+		"01234567890abc", // >11 chars
+		"!!!!",           // invalid chars
+		"-underscore-",   // invalid chars
+		"abc_def",        // underscore invalid for base62
+		"0123456789ABCD", // 14 chars
 	}
-	for _, in := range inputs {
-		a := FallbackFromString(in)
-		b := FallbackFromString(in)
-		if a.raw != b.raw || a.String() != b.String() {
-			t.Fatalf("fallback mismatch for %q: %+v vs %+v", in, a, b)
+	for _, s := range invalid {
+		if _, ok := ParseCartId(s); ok {
+			t.Fatalf("expected parse failure for %q", s)
 		}
 	}
-	// Distinct inputs should almost always differ; sample check
-	a := FallbackFromString("distinct-A")
-	b := FallbackFromString("distinct-B")
-	if a.raw == b.raw {
-		t.Fatalf("unexpected identical fallback hashes for distinct inputs")
-	}
 }
 
-// TestCanonicalizeIncomingBehavior covers main control flow branches.
-func TestCanonicalizeIncomingBehavior(t *testing.T) {
-	// Empty => new id
-	id1, generated, err := CanonicalizeIncoming("")
-	if err != nil || !generated || id1.IsZero() {
-		t.Fatalf("CanonicalizeIncoming empty failed: id=%v gen=%v err=%v", id1, generated, err)
+// TestMustParseCartIdPanics verifies panic behavior for invalid input.
+func TestMustParseCartIdPanics(t *testing.T) {
+	defer func() {
+		if r := recover(); r == nil {
+			t.Fatalf("expected panic for invalid MustParseCartId input")
 		}
+	}()
+	_ = MustParseCartId("not*base62")
+}
 
-	// Valid base62 => parse; no generation
-	id2, gen2, err := CanonicalizeIncoming(id1.String())
-	if err != nil || gen2 || id2.raw != id1.raw {
-		t.Fatalf("CanonicalizeIncoming parse mismatch: id2=%v gen2=%v err=%v", id2, gen2, err)
+// TestJSONMarshalUnmarshalCartId verifies JSON round trip.
+func TestJSONMarshalUnmarshalCartId(t *testing.T) {
+	id := MustNewCartId()
+	data, err := json.Marshal(struct {
+		Cart CartId `json:"cart"`
+	}{Cart: id})
+	if err != nil {
+		t.Fatalf("marshal error: %v", err)
 	}
-
-	// Legacy-like random containing invalid chars -> fallback
-	fallbackInput := "legacy\x00\x00padding"
-	id3, gen3, err := CanonicalizeIncoming(fallbackInput)
-	if err != nil || gen3 {
-		t.Fatalf("CanonicalizeIncoming fallback unexpected: id3=%v gen3=%v err=%v", id3, gen3, err)
+	var out struct {
+		Cart CartId `json:"cart"`
 	}
-
-	// Deterministic fallback
-	id4, _, _ := CanonicalizeIncoming(fallbackInput)
-	if id3.raw != id4.raw {
-		t.Fatalf("fallback canonicalization not deterministic")
+	if err := json.Unmarshal(data, &out); err != nil {
+		t.Fatalf("unmarshal error: %v", err)
+	}
+	if out.Cart != id {
+		t.Fatalf("JSON round trip mismatch: have %d got %d", id, out.Cart)
 	}
 }
 
-// TestUpgradeLegacyCartId ensures mapping of old CartId is stable.
-func TestUpgradeLegacyCartId(t *testing.T) {
-	var legacy CartId
-	copy(legacy[:], []byte("legacy-123456789")) // 15 bytes + padding
-	up1 := UpgradeLegacyCartId(legacy)
-	up2 := UpgradeLegacyCartId(legacy)
-	if up1.raw != up2.raw {
-		t.Fatalf("UpgradeLegacyCartId not deterministic: %v vs %v", up1, up2)
+// TestBase62LengthBound checks worst-case length (near max uint64).
+func TestBase62LengthBound(t *testing.T) {
+	// Largest uint64
+	const maxU64 = ^uint64(0)
+	s := encodeBase62(maxU64)
+	if len(s) > 11 {
+		t.Fatalf("max uint64 encoded length > 11: %d (%s)", len(s), s)
 	}
-	if up1.String() != up2.String() {
-		t.Fatalf("UpgradeLegacyCartId string mismatch: %s vs %s", up1, up2)
+	dec, ok := decodeBase62(s)
+	if !ok || dec != maxU64 {
+		t.Fatalf("decode failed for max uint64: ok=%v dec=%d want=%d", ok, dec, maxU64)
 	}
 }
 
-// BenchmarkNewCartID gives a rough idea of generation cost.
-func BenchmarkNewCartID(b *testing.B) {
+// TestZeroEncoding ensures zero value encodes to "0" and parses back.
+func TestZeroEncoding(t *testing.T) {
+	if s := encodeBase62(0); s != "0" {
+		t.Fatalf("encodeBase62(0) expected '0', got %q", s)
+	}
+	v, ok := decodeBase62("0")
+	if !ok || v != 0 {
+		t.Fatalf("decodeBase62('0') failed: ok=%v v=%d", ok, v)
+	}
+	if _, ok := ParseCartId("0"); !ok {
+		t.Fatalf("ParseCartId(\"0\") should succeed")
+	}
+}
+
+// TestSequentialParse ensures sequentially generated ids parse correctly.
+func TestSequentialParse(t *testing.T) {
+	for i := 0; i < 1000; i++ {
+		id := MustNewCartId()
+		txt := id.String()
+		parsed, ok := ParseCartId(txt)
+		if !ok || parsed != id {
+			t.Fatalf("sequential parse mismatch: idx=%d orig=%d parsed=%d txt=%s", i, id, parsed, txt)
+		}
+	}
+}
+
+// BenchmarkNewCartId measures generation performance.
+func BenchmarkNewCartId(b *testing.B) {
 	for i := 0; i < b.N; i++ {
-		if _, err := NewCartID(); err != nil {
-			b.Fatalf("error: %v", err)
+		if _, err := NewCartId(); err != nil {
+			b.Fatalf("NewCartId error: %v", err)
 		}
 	}
 }
 
-// BenchmarkEncodeBase62 measures encode speed in isolation.
+// BenchmarkEncodeBase62 measures encoding performance.
 func BenchmarkEncodeBase62(b *testing.B) {
-	// Random sample of values
+	// Precompute sample values
 	samples := make([]uint64, 1024)
 	for i := range samples {
-		var buf [8]byte
-		if _, err := rand.Read(buf[:]); err != nil {
-			b.Fatalf("rand: %v", err)
-		}
-		samples[i] = binary.BigEndian.Uint64(buf[:])
+		// Spread bits without crypto randomness overhead
+		samples[i] = (uint64(i) << 53) ^ (uint64(i) * 0x9E3779B185EBCA87)
 	}
 	b.ResetTimer()
 	var sink string
@@ -172,88 +159,27 @@ func BenchmarkEncodeBase62(b *testing.B) {
 	_ = sink
 }
 
-// BenchmarkDecodeBase62 measures decode speed.
+// BenchmarkDecodeBase62 measures decoding performance.
 func BenchmarkDecodeBase62(b *testing.B) {
-	// Pre-encode
 	encoded := make([]string, 1024)
 	for i := range encoded {
-		encoded[i] = encodeBase62(uint64(i)<<32 | uint64(i))
+		encoded[i] = encodeBase62((uint64(i) << 32) | uint64(i))
 	}
 	b.ResetTimer()
 	var sum uint64
 	for i := 0; i < b.N; i++ {
 		v, ok := decodeBase62(encoded[i%len(encoded)])
 		if !ok {
-			b.Fatalf("decode failed")
+			b.Fatalf("decode failure for %s", encoded[i%len(encoded)])
 		}
 		sum ^= v
 	}
 	_ = sum
 }
 
-// TestLookupNDeterminism (ring integration smoke test) ensures LookupN
-// returns distinct hosts and stable ordering for a fixed ring.
-func TestLookupNDeterminism(t *testing.T) {
-	rb := NewRingBuilder().WithEpoch(1).WithVnodesPerHost(8).WithHosts([]string{"a", "b", "c"})
-	ring := rb.Build()
-	if ring.Empty() {
-		t.Fatalf("expected non-empty ring")
-	}
-	id := MustNewCartID()
-	owners1 := ring.LookupN(id.Raw(), 3)
-	owners2 := ring.LookupN(id.Raw(), 3)
-	if len(owners1) != len(owners2) {
-		t.Fatalf("LookupN length mismatch")
-	}
-	for i := range owners1 {
-		if owners1[i].Host != owners2[i].Host {
-			t.Fatalf("LookupN ordering instability at %d: %v vs %v", i, owners1[i], owners2[i])
-		}
-	}
-	// Distinct host constraint
-	seen := map[string]struct{}{}
-	for _, v := range owners1 {
-		if _, ok := seen[v.Host]; ok {
-			t.Fatalf("duplicate host in LookupN result: %v", owners1)
-		}
-		seen[v.Host] = struct{}{}
-	}
-}
-
-// TestRingFingerprintChanges ensures fingerprint updates with membership changes.
-func TestRingFingerprintChanges(t *testing.T) {
-	b1 := NewRingBuilder().WithEpoch(1).WithHosts([]string{"node1", "node2"})
-	r1 := b1.Build()
-	b2 := NewRingBuilder().WithEpoch(2).WithHosts([]string{"node1", "node2", "node3"})
-	r2 := b2.Build()
-	if r1.Fingerprint() == r2.Fingerprint() {
-		t.Fatalf("expected differing fingerprints after host set change")
-	}
-}
-
-// TestRingDiffHosts verifies added/removed host detection.
-func TestRingDiffHosts(t *testing.T) {
-	r1 := NewRingBuilder().WithEpoch(1).WithHosts([]string{"a", "b"}).Build()
-	r2 := NewRingBuilder().WithEpoch(2).WithHosts([]string{"b", "c"}).Build()
-	added, removed := r1.DiffHosts(r2)
-	if fmt.Sprintf("%v", added) != "[c]" {
-		t.Fatalf("expected added [c], got %v", added)
-	}
-	if fmt.Sprintf("%v", removed) != "[a]" {
-		t.Fatalf("expected removed [a], got %v", removed)
-	}
-}
-
-// TestRingLookupConsistency ensures direct Lookup and LookupID are aligned.
-func TestRingLookupConsistency(t *testing.T) {
-	ring := NewRingBuilder().WithEpoch(1).WithHosts([]string{"alpha", "beta"}).WithVnodesPerHost(4).Build()
-	id, _ := ParseCartID("1")
-	if id.IsZero() {
-		t.Fatalf("expected parsed id non-zero")
-	}
-	v1 := ring.Lookup(id.Raw())
-	v2 := ring.LookupID(id)
-	if v1.Host != v2.Host || v1.Hash != v2.Hash {
-		t.Fatalf("Lookup vs LookupID mismatch: %+v vs %+v", v1, v2)
-	}
+// ExampleCartIdString documents usage of CartId string form.
+func ExampleCartId_string() {
+	id := MustNewCartId()
+	fmt.Println(len(id.String()) <= 11) // outputs true
+	// Output: true
 }

cart_state_mapper.go

@@ -1,212 +0,0 @@
-package main
-
-import (
-	messages "git.tornberg.me/go-cart-actor/proto"
-)
-
-// cart_state_mapper.go
-//
-// Utilities to translate between internal CartGrain state and the gRPC
-// (typed) protobuf representation CartState. This replaces the previous
-// JSON blob framing and enables type-safe replies over gRPC, as well as
-// internal reuse for HTTP handlers without an extra marshal / unmarshal
-// hop (you can marshal CartState directly for JSON responses if desired).
-//
-// Only the one‑way mapping (CartGrain -> CartState) is strictly required
-// for mutation / state replies. A reverse helper is included in case
-// future features (e.g. snapshot import, replay, or migration) need it.
-
-// ToCartState converts the in‑memory CartGrain into a protobuf CartState.
-func ToCartState(c *CartGrain) *messages.CartState {
-	if c == nil {
-		return nil
-	}
-
-	items := make([]*messages.CartItemState, 0, len(c.Items))
-	for _, it := range c.Items {
-		if it == nil {
-			continue
-		}
-		itemDiscountPerUnit := max(0, it.OrgPrice-it.Price)
-		itemTotalDiscount := itemDiscountPerUnit * int64(it.Quantity)
-
-		items = append(items, &messages.CartItemState{
-			Id:            int64(it.Id),
-			ItemId:        int64(it.ItemId),
-			Sku:           it.Sku,
-			Name:          it.Name,
-			Price:         it.Price,
-			Qty:           int32(it.Quantity),
-			TotalPrice:    it.TotalPrice,
-			TotalTax:      it.TotalTax,
-			OrgPrice:      it.OrgPrice,
-			TaxRate:       int32(it.TaxRate),
-			TotalDiscount: itemTotalDiscount,
-			Brand:         it.Brand,
-			Category:      it.Category,
-			Category2:     it.Category2,
-			Category3:     it.Category3,
-			Category4:     it.Category4,
-			Category5:     it.Category5,
-			Image:         it.Image,
-			Type:          it.ArticleType,
-			SellerId:      it.SellerId,
-			SellerName:    it.SellerName,
-			Disclaimer:    it.Disclaimer,
-			Outlet:        deref(it.Outlet),
-			StoreId:       deref(it.StoreId),
-			Stock:         int32(it.Stock),
-		})
-	}
-
-	deliveries := make([]*messages.DeliveryState, 0, len(c.Deliveries))
-	for _, d := range c.Deliveries {
-		if d == nil {
-			continue
-		}
-		itemIds := make([]int64, 0, len(d.Items))
-		for _, id := range d.Items {
-			itemIds = append(itemIds, int64(id))
-		}
-		var pp *messages.PickupPoint
-		if d.PickupPoint != nil {
-			// Copy to avoid accidental shared mutation (proto points are fine but explicit).
-			pp = &messages.PickupPoint{
-				Id:      d.PickupPoint.Id,
-				Name:    d.PickupPoint.Name,
-				Address: d.PickupPoint.Address,
-				City:    d.PickupPoint.City,
-				Zip:     d.PickupPoint.Zip,
-				Country: d.PickupPoint.Country,
-			}
-		}
-		deliveries = append(deliveries, &messages.DeliveryState{
-			Id:          int64(d.Id),
-			Provider:    d.Provider,
-			Price:       d.Price,
-			Items:       itemIds,
-			PickupPoint: pp,
-		})
-	}
-
-	return &messages.CartState{
-		Id:                c.Id.String(),
-		Items:             items,
-		TotalPrice:        c.TotalPrice,
-		TotalTax:          c.TotalTax,
-		TotalDiscount:     c.TotalDiscount,
-		Deliveries:        deliveries,
-		PaymentInProgress: c.PaymentInProgress,
-		OrderReference:    c.OrderReference,
-		PaymentStatus:     c.PaymentStatus,
-	}
-}
-
-// FromCartState merges a protobuf CartState into an existing CartGrain.
-// This is optional and primarily useful for snapshot import or testing.
-func FromCartState(cs *messages.CartState, g *CartGrain) *CartGrain {
-	if cs == nil {
-		return g
-	}
-	if g == nil {
-		g = &CartGrain{}
-	}
-	g.Id = ToCartId(cs.Id)
-	g.TotalPrice = cs.TotalPrice
-	g.TotalTax = cs.TotalTax
-	g.TotalDiscount = cs.TotalDiscount
-	g.PaymentInProgress = cs.PaymentInProgress
-	g.OrderReference = cs.OrderReference
-	g.PaymentStatus = cs.PaymentStatus
-
-	// Items
-	g.Items = g.Items[:0]
-	for _, it := range cs.Items {
-		if it == nil {
-			continue
-		}
-		outlet := toPtr(it.Outlet)
-		storeId := toPtr(it.StoreId)
-		g.Items = append(g.Items, &CartItem{
-			Id:          int(it.Id),
-			ItemId:      int(it.ItemId),
-			Sku:         it.Sku,
-			Name:        it.Name,
-			Price:       it.Price,
-			Quantity:    int(it.Qty),
-			TotalPrice:  it.TotalPrice,
-			TotalTax:    it.TotalTax,
-			OrgPrice:    it.OrgPrice,
-			TaxRate:     int(it.TaxRate),
-			Brand:       it.Brand,
-			Category:    it.Category,
-			Category2:   it.Category2,
-			Category3:   it.Category3,
-			Category4:   it.Category4,
-			Category5:   it.Category5,
-			Image:       it.Image,
-			ArticleType: it.Type,
-			SellerId:    it.SellerId,
-			SellerName:  it.SellerName,
-			Disclaimer:  it.Disclaimer,
-			Outlet:      outlet,
-			StoreId:     storeId,
-			Stock:       StockStatus(it.Stock),
-			// Tax, TaxRate already set via Price / Totals if needed
-		})
-		if it.Id > int64(g.lastItemId) {
-			g.lastItemId = int(it.Id)
-		}
-	}
-
-	// Deliveries
-	g.Deliveries = g.Deliveries[:0]
-	for _, d := range cs.Deliveries {
-		if d == nil {
-			continue
-		}
-
-		intIds := make([]int, 0, len(d.Items))
-		for _, id := range d.Items {
-			intIds = append(intIds, int(id))
-		}
-		var pp *messages.PickupPoint
-		if d.PickupPoint != nil {
-			pp = &messages.PickupPoint{
-				Id:      d.PickupPoint.Id,
-				Name:    d.PickupPoint.Name,
-				Address: d.PickupPoint.Address,
-				City:    d.PickupPoint.City,
-				Zip:     d.PickupPoint.Zip,
-				Country: d.PickupPoint.Country,
-			}
-		}
-		g.Deliveries = append(g.Deliveries, &CartDelivery{
-			Id:          int(d.Id),
-			Provider:    d.Provider,
-			Price:       d.Price,
-			Items:       intIds,
-			PickupPoint: pp,
-		})
-		if d.Id > int64(g.lastDeliveryId) {
-			g.lastDeliveryId = int(d.Id)
-		}
-	}
-
-	return g
-}
-
-// Helper to safely de-reference optional string pointers to value or "".
-func deref(p *string) string {
-	if p == nil {
-		return ""
-	}
-	return *p
-}
-
-func toPtr(s string) *string {
-	if s == "" {
-		return nil
-	}
-	return &s
-}

deployment/deployment.yaml

@@ -15,7 +15,7 @@ metadata:
     arch: amd64
   name: cart-actor-x86
 spec:
-    replicas: 0
+  replicas: 3
   selector:
     matchLabels:
       app: cart-actor
@@ -113,7 +113,7 @@ metadata:
     arch: arm64
   name: cart-actor-arm64
 spec:
-    replicas: 3
+  replicas: 0
   selector:
     matchLabels:
       app: cart-actor
@@ -228,10 +228,10 @@ metadata:
   name: cart-ingress
   annotations:
     cert-manager.io/cluster-issuer: letsencrypt-prod
-        # nginx.ingress.kubernetes.io/affinity: "cookie"
-        # nginx.ingress.kubernetes.io/session-cookie-name: "cart-affinity"
-        # nginx.ingress.kubernetes.io/session-cookie-expires: "172800"
-        # nginx.ingress.kubernetes.io/session-cookie-max-age: "172800"
+    nginx.ingress.kubernetes.io/affinity: "cookie"
+    nginx.ingress.kubernetes.io/session-cookie-name: "cart-affinity"
+    nginx.ingress.kubernetes.io/session-cookie-expires: "172800"
+    nginx.ingress.kubernetes.io/session-cookie-max-age: "172800"
     nginx.ingress.kubernetes.io/proxy-body-size: 4m
 spec:
   ingressClassName: nginx

discarded-host.go

@@ -1,84 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"log"
-	"net"
-	"sync"
-	"time"
-)
-
-type DiscardedHost struct {
-	Host  string
-	Tries int
-}
-
-type DiscardedHostHandler struct {
-	mu           sync.RWMutex
-	port         int
-	hosts        []*DiscardedHost
-	onConnection *func(string)
-}
-
-func (d *DiscardedHostHandler) run() {
-	for range time.Tick(time.Second) {
-		d.mu.RLock()
-		lst := make([]*DiscardedHost, 0, len(d.hosts))
-		for _, host := range d.hosts {
-			if host.Tries >= 0 && host.Tries < 5 {
-				go d.testConnection(host)
-				lst = append(lst, host)
-			} else {
-				if host.Tries > 0 {
-					log.Printf("Host %s discarded after %d tries", host.Host, host.Tries)
-				}
-			}
-		}
-		d.mu.RUnlock()
-		d.mu.Lock()
-		d.hosts = lst
-		d.mu.Unlock()
-	}
-
-}
-
-func (d *DiscardedHostHandler) testConnection(host *DiscardedHost) {
-	addr := fmt.Sprintf("%s:%d", host.Host, d.port)
-	conn, err := net.Dial("tcp", addr)
-
-	if err != nil {
-		host.Tries++
-		if host.Tries >= 5 {
-			// Exceeded retry threshold; will be dropped by run loop.
-		}
-	} else {
-		conn.Close()
-		if d.onConnection != nil {
-			fn := *d.onConnection
-			fn(host.Host)
-		}
-	}
-}
-
-func NewDiscardedHostHandler(port int) *DiscardedHostHandler {
-	ret := &DiscardedHostHandler{
-		hosts: make([]*DiscardedHost, 0),
-		port:  port,
-	}
-	go ret.run()
-	return ret
-}
-
-func (d *DiscardedHostHandler) SetReconnectHandler(fn func(string)) {
-	d.onConnection = &fn
-}
-
-func (d *DiscardedHostHandler) AppendHost(host string) {
-	d.mu.Lock()
-	defer d.mu.Unlock()
-	log.Printf("Adding host %s to retry list", host)
-	d.hosts = append(d.hosts, &DiscardedHost{
-		Host:  host,
-		Tries: 0,
-	})
-}

disk-storage.go

@@ -2,69 +2,72 @@ package main
 
 import (
 	"encoding/gob"
-	"fmt"
-	"os"
 	"time"
 )
 
+func init() {
+
+	gob.Register(map[uint64]int64{})
+}
+
 type DiskStorage struct {
 	stateFile string
-	lastSave  int64
-	LastSaves map[CartId]int64
+	lastSave  time.Time
+	LastSaves map[uint64]time.Time
 }
 
 func NewDiskStorage(stateFile string) (*DiskStorage, error) {
 	ret := &DiskStorage{
 		stateFile: stateFile,
-		LastSaves: make(map[CartId]int64),
+		LastSaves: make(map[uint64]time.Time),
 	}
-	err := ret.loadState()
-	return ret, err
+	//err := ret.loadState()
+	return ret, nil
 }
 
-func saveMessages(_ interface{}, _ CartId) error {
-	// No-op: legacy event log persistence removed in oneof refactor.
-	return nil
-}
+// func saveMessages(_ interface{}, _ CartId) error {
+// 	// No-op: legacy event log persistence removed in oneof refactor.
+// 	return nil
+// }
 
-func getCartPath(id string) string {
-	return fmt.Sprintf("data/%s.prot", id)
-}
+// func getCartPath(id string) string {
+// 	return fmt.Sprintf("data/%s.prot", id)
+// }
 
-func loadMessages(_ Grain, _ CartId) error {
-	// No-op: legacy replay removed in oneof refactor.
-	return nil
-}
+// func loadMessages(_ Grain, _ CartId) error {
+// 	// No-op: legacy replay removed in oneof refactor.
+// 	return nil
+// }
 
-func (s *DiskStorage) saveState() error {
-	tmpFile := s.stateFile + "_tmp"
-	file, err := os.Create(tmpFile)
-	if err != nil {
-		return err
-	}
-	defer file.Close()
-	err = gob.NewEncoder(file).Encode(s.LastSaves)
-	if err != nil {
-		return err
-	}
-	os.Remove(s.stateFile + ".bak")
-	os.Rename(s.stateFile, s.stateFile+".bak")
-	return os.Rename(tmpFile, s.stateFile)
-}
+// func (s *DiskStorage) saveState() error {
+// 	tmpFile := s.stateFile + "_tmp"
+// 	file, err := os.Create(tmpFile)
+// 	if err != nil {
+// 		return err
+// 	}
+// 	defer file.Close()
+// 	err = gob.NewEncoder(file).Encode(s.LastSaves)
+// 	if err != nil {
+// 		return err
+// 	}
+// 	os.Remove(s.stateFile + ".bak")
+// 	os.Rename(s.stateFile, s.stateFile+".bak")
+// 	return os.Rename(tmpFile, s.stateFile)
+// }
 
-func (s *DiskStorage) loadState() error {
-	file, err := os.Open(s.stateFile)
-	if err != nil {
-		return err
-	}
-	defer file.Close()
-	return gob.NewDecoder(file).Decode(&s.LastSaves)
-}
+// func (s *DiskStorage) loadState() error {
+// 	file, err := os.Open(s.stateFile)
+// 	if err != nil {
+// 		return err
+// 	}
+// 	defer file.Close()
+// 	return gob.NewDecoder(file).Decode(&s.LastSaves)
+// }
 
 func (s *DiskStorage) Store(id CartId, _ *CartGrain) error {
 	// With the removal of the legacy message log, we only update the timestamp.
-	ts := time.Now().Unix()
-	s.LastSaves[id] = ts
+	ts := time.Now()
+	s.LastSaves[uint64(id)] = ts
 	s.lastSave = ts
 	return nil
 }

event_log.go

@@ -0,0 +1,288 @@
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
+	"sync"
+	"time"
+
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+/*
+event_log.go
+
+Append-only cart event log (per cart id) with replay + metrics.
+
+Rationale:
+  - Enables recovery of in-memory cart state after process restarts or TTL eviction.
+  - Provides a chronological mutation trail for auditing / debugging.
+  - Avoids write amplification of full snapshots on every mutation.
+
+Format:
+  One JSON object per line:
+   {
+     "ts": 1700000000,
+     "type": "AddRequest",
+     "payload": { ... mutation fields ... }
+   }
+
+Concurrency:
+  - Appends: synchronized per-cart via an in-process mutex map to avoid partial writes.
+  - Replay: sequential read of entire file; mutations applied in order.
+
+Usage Integration (to be wired by caller):
+  1. After successful mutation application (non-replay), invoke:
+        AppendCartEvent(grain.GetId(), mutation)
+  2. During grain spawn, call:
+        ReplayCartEvents(grain, grain.GetId())
+     BEFORE serving requests, so state is reconstructed.
+
+Metrics:
+  - cart_event_log_appends_total
+  - cart_event_log_replay_total
+  - cart_event_log_replay_failures_total
+  - cart_event_log_bytes_written_total
+  - cart_event_log_files_existing (gauge)
+  - cart_event_log_last_append_unix (gauge)
+  - cart_event_log_replay_duration_seconds (histogram)
+
+Rotation / Compaction:
+  - Not implemented. If needed, implement size checks and snapshot+truncate later.
+
+Caveats:
+  - Mutation schema changes may break replay unless backward-compatible.
+  - Missing / unknown event types are skipped (metric incremented).
+  - If a mutation fails during replay, replay continues (logged + metric).
+
+*/
+
+var (
+	eventLogDir = "data"
+
+	eventAppendsTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cart_event_log_appends_total",
+		Help: "Total number of cart mutation events appended to event logs.",
+	})
+	eventReplayTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cart_event_log_replay_total",
+		Help: "Total number of successful event log replays (per cart).",
+	})
+	eventReplayFailuresTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cart_event_log_replay_failures_total",
+		Help: "Total number of failed event log replay operations.",
+	})
+	eventBytesWrittenTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cart_event_log_bytes_written_total",
+		Help: "Cumulative number of bytes written to all cart event logs.",
+	})
+	eventFilesExisting = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "cart_event_log_files_existing",
+		Help: "Number of cart event log files currently existing on disk.",
+	})
+	eventLastAppendUnix = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "cart_event_log_last_append_unix",
+		Help: "Unix timestamp of the last append to any cart event log.",
+	})
+	eventReplayDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+		Name:    "cart_event_log_replay_duration_seconds",
+		Help:    "Duration of replay operations per cart in seconds.",
+		Buckets: prometheus.DefBuckets,
+	})
+	eventUnknownTypesTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cart_event_log_unknown_types_total",
+		Help: "Total number of unknown event types encountered during replay (skipped).",
+	})
+	eventMutationErrorsTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cart_event_log_mutation_errors_total",
+		Help: "Total number of errors applying mutation events during replay.",
+	})
+)
+
+type cartEventRecord struct {
+	Timestamp int64       `json:"ts"`
+	Type      string      `json:"type"`
+	Payload   interface{} `json:"payload"`
+}
+
+// registry of supported mutation payload type constructors
+var eventTypeFactories = map[string]func() interface{}{
+	"AddRequest":         func() interface{} { return &messages.AddRequest{} },
+	"AddItem":            func() interface{} { return &messages.AddItem{} },
+	"RemoveItem":         func() interface{} { return &messages.RemoveItem{} },
+	"RemoveDelivery":     func() interface{} { return &messages.RemoveDelivery{} },
+	"ChangeQuantity":     func() interface{} { return &messages.ChangeQuantity{} },
+	"SetDelivery":        func() interface{} { return &messages.SetDelivery{} },
+	"SetPickupPoint":     func() interface{} { return &messages.SetPickupPoint{} },
+	"SetCartRequest":     func() interface{} { return &messages.SetCartRequest{} },
+	"OrderCreated":       func() interface{} { return &messages.OrderCreated{} },
+	"InitializeCheckout": func() interface{} { return &messages.InitializeCheckout{} },
+}
+
+// Per-cart mutexes to serialize append operations (avoid partial overlapping writes)
+var (
+	eventLogMu sync.Map // map[string]*sync.Mutex
+)
+
+// getCartEventMutex returns a mutex for a specific cart id string.
+func getCartEventMutex(id string) *sync.Mutex {
+	if v, ok := eventLogMu.Load(id); ok {
+		return v.(*sync.Mutex)
+	}
+	m := &sync.Mutex{}
+	actual, _ := eventLogMu.LoadOrStore(id, m)
+	return actual.(*sync.Mutex)
+}
+
+// EventLogPath returns the path to the cart's event log file.
+func EventLogPath(id CartId) string {
+	return filepath.Join(eventLogDir, fmt.Sprintf("%s.events.log", id.String()))
+}
+
+// EnsureEventLogDirectory ensures base directory exists and updates gauge.
+func EnsureEventLogDirectory() error {
+	if _, err := os.Stat(eventLogDir); errors.Is(err, os.ErrNotExist) {
+		if err2 := os.MkdirAll(eventLogDir, 0755); err2 != nil {
+			return err2
+		}
+	}
+	// Update files existing gauge (approximate; counts matching *.events.log)
+	pattern := filepath.Join(eventLogDir, "*.events.log")
+	matches, _ := filepath.Glob(pattern)
+	eventFilesExisting.Set(float64(len(matches)))
+	return nil
+}
+
+// AppendCartEvent appends a mutation event to the cart's log (JSON line).
+func AppendCartEvent(id CartId, mutation interface{}) error {
+	if mutation == nil {
+		return errors.New("nil mutation cannot be logged")
+	}
+	if err := EnsureEventLogDirectory(); err != nil {
+		return err
+	}
+
+	typ := mutationTypeName(mutation)
+	rec := cartEventRecord{
+		Timestamp: time.Now().Unix(),
+		Type:      typ,
+		Payload:   mutation,
+	}
+	lineBytes, err := json.Marshal(rec)
+	if err != nil {
+		return fmt.Errorf("marshal event: %w", err)
+	}
+	lineBytes = append(lineBytes, '\n')
+
+	path := EventLogPath(id)
+	mtx := getCartEventMutex(id.String())
+	mtx.Lock()
+	defer mtx.Unlock()
+
+	fh, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
+	if err != nil {
+		return fmt.Errorf("open event log: %w", err)
+	}
+	defer fh.Close()
+
+	n, werr := fh.Write(lineBytes)
+	if werr != nil {
+		return fmt.Errorf("write event log: %w", werr)
+	}
+
+	eventAppendsTotal.Inc()
+	eventBytesWrittenTotal.Add(float64(n))
+	eventLastAppendUnix.Set(float64(rec.Timestamp))
+	return nil
+}
+
+// ReplayCartEvents replays an existing cart's event log into the provided grain.
+// It applies mutation payloads in order, skipping unknown types.
+func ReplayCartEvents(grain *CartGrain, id CartId) error {
+	start := time.Now()
+	path := EventLogPath(id)
+	if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
+		// No log -> nothing to replay
+		return nil
+	}
+
+	fh, err := os.Open(path)
+	if err != nil {
+		eventReplayFailuresTotal.Inc()
+		return fmt.Errorf("open replay file: %w", err)
+	}
+	defer fh.Close()
+
+	scanner := bufio.NewScanner(fh)
+	// Increase buffer in case of large payloads
+	const maxLine = 256 * 1024
+	buf := make([]byte, 0, 64*1024)
+	scanner.Buffer(buf, maxLine)
+
+	for scanner.Scan() {
+		line := scanner.Bytes()
+		var raw struct {
+			Timestamp time.Time       `json:"ts"`
+			Type      string          `json:"type"`
+			Payload   json.RawMessage `json:"payload"`
+		}
+		if err := json.Unmarshal(line, &raw); err != nil {
+			eventReplayFailuresTotal.Inc()
+			continue // skip malformed line
+		}
+		factory, ok := eventTypeFactories[raw.Type]
+		if !ok {
+			eventUnknownTypesTotal.Inc()
+			continue // skip unknown mutation type
+		}
+		instance := factory()
+		if err := json.Unmarshal(raw.Payload, instance); err != nil {
+			eventMutationErrorsTotal.Inc()
+			continue
+		}
+		// Apply mutation directly using internal registration (bypass AppendCartEvent recursion).
+		if _, applyErr := ApplyRegistered(grain, instance); applyErr != nil {
+			eventMutationErrorsTotal.Inc()
+			continue
+		} else {
+			// Update lastChange to the timestamp of this event (sliding inactivity window support).
+			grain.lastChange = raw.Timestamp
+		}
+	}
+	if serr := scanner.Err(); serr != nil {
+		eventReplayFailuresTotal.Inc()
+		return fmt.Errorf("scanner error: %w", serr)
+	}
+
+	eventReplayTotal.Inc()
+	eventReplayDuration.Observe(time.Since(start).Seconds())
+	return nil
+}
+
+// mutationTypeName returns the short struct name for a mutation (pointer aware).
+func mutationTypeName(v interface{}) string {
+	if v == nil {
+		return "nil"
+	}
+	t := reflect.TypeOf(v)
+	if t.Kind() == reflect.Ptr {
+		t = t.Elem()
+	}
+	return t.Name()
+}
+
+/*
+Future enhancements:
+  - Compression: gzip large (> N events) logs to reduce disk usage.
+  - Compaction: periodic snapshot + truncate old events to bound replay latency.
+  - Checkpoint events: inject cart state snapshots every M mutations.
+  - Integrity: add checksum per line for corruption detection.
+  - Multi-writer safety across processes (currently only safe within one process).
+*/

grafana_dashboard_cart.json

@@ -0,0 +1,327 @@
+{
+  "uid": "cart-actors",
+  "title": "Cart Actor Cluster",
+  "timezone": "browser",
+  "refresh": "30s",
+  "schemaVersion": 38,
+  "version": 1,
+  "editable": true,
+  "graphTooltip": 0,
+  "panels": [
+    {
+      "type": "row",
+      "title": "Overview",
+      "gridPos": { "x": 0, "y": 0, "w": 24, "h": 1 },
+      "id": 1,
+      "collapsed": false
+    },
+    {
+      "type": "stat",
+      "title": "Active Grains",
+      "id": 2,
+      "gridPos": { "x": 0, "y": 1, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "cart_active_grains" }
+      ],
+      "options": {
+        "colorMode": "value",
+        "graphMode": "none",
+        "justifyMode": "center",
+        "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false }
+      }
+    },
+    {
+      "type": "stat",
+      "title": "Grains In Pool",
+      "id": 3,
+      "gridPos": { "x": 6, "y": 1, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "cart_grains_in_pool" }
+      ],
+      "options": {
+        "colorMode": "value",
+        "graphMode": "none",
+        "justifyMode": "center",
+        "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false }
+      }
+    },
+    {
+      "type": "stat",
+      "title": "Pool Usage %",
+      "id": 4,
+      "gridPos": { "x": 12, "y": 1, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "cart_grain_pool_usage * 100" }
+      ],
+      "units": "percent",
+      "options": {
+        "colorMode": "value",
+        "graphMode": "none",
+        "justifyMode": "center",
+        "reduceOptions": { "calcs": ["lastNotNull"] }
+      }
+    },
+    {
+      "type": "stat",
+      "title": "Connected Remotes",
+      "id": 5,
+      "gridPos": { "x": 18, "y": 1, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "connected_remotes" }
+      ],
+      "options": {
+        "colorMode": "value",
+        "graphMode": "none",
+        "justifyMode": "center",
+        "reduceOptions": { "calcs": ["lastNotNull"] }
+      }
+    },
+
+    {
+      "type": "row",
+      "title": "Mutations",
+      "gridPos": { "x": 0, "y": 5, "w": 24, "h": 1 },
+      "id": 6,
+      "collapsed": false
+    },
+    {
+      "type": "timeseries",
+      "title": "Mutation Rate (1m)",
+      "id": 7,
+      "gridPos": { "x": 0, "y": 6, "w": 12, "h": 8 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "rate(cart_mutations_total[1m])", "legendFormat": "mutations/s" },
+        { "refId": "B", "expr": "rate(cart_mutation_failures_total[1m])", "legendFormat": "failures/s" }
+      ],
+      "fieldConfig": { "defaults": { "unit": "ops" } }
+    },
+    {
+      "type": "stat",
+      "title": "Failure % (5m)",
+      "id": 8,
+      "gridPos": { "x": 12, "y": 6, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "100 * (increase(cart_mutation_failures_total[5m]) / clamp_max(increase(cart_mutations_total[5m]), 1))"
+        }
+      ],
+      "options": {
+        "colorMode": "value",
+        "graphMode": "none",
+        "justifyMode": "center",
+        "reduceOptions": { "calcs": ["lastNotNull"] }
+      }
+    },
+    {
+      "type": "timeseries",
+      "title": "Mutation Latency Quantiles",
+      "id": 9,
+      "gridPos": { "x": 18, "y": 6, "w": 6, "h": 8 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "histogram_quantile(0.50, sum(rate(cart_mutation_latency_seconds_bucket[5m])) by (le))",
+          "legendFormat": "p50"
+        },
+        {
+          "refId": "B",
+          "expr": "histogram_quantile(0.90, sum(rate(cart_mutation_latency_seconds_bucket[5m])) by (le))",
+          "legendFormat": "p90"
+        },
+        {
+          "refId": "C",
+          "expr": "histogram_quantile(0.99, sum(rate(cart_mutation_latency_seconds_bucket[5m])) by (le))",
+          "legendFormat": "p99"
+        }
+      ],
+      "fieldConfig": { "defaults": { "unit": "s" } }
+    },
+
+    {
+      "type": "row",
+      "title": "Event Log",
+      "gridPos": { "x": 0, "y": 14, "w": 24, "h": 1 },
+      "id": 10,
+      "collapsed": false
+    },
+    {
+      "type": "timeseries",
+      "title": "Event Append Rate (5m)",
+      "id": 11,
+      "gridPos": { "x": 0, "y": 15, "w": 8, "h": 6 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "rate(cart_event_log_appends_total[5m])", "legendFormat": "appends/s" }
+      ]
+    },
+    {
+      "type": "timeseries",
+      "title": "Event Bytes Written Rate (5m)",
+      "id": 12,
+      "gridPos": { "x": 8, "y": 15, "w": 8, "h": 6 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "rate(cart_event_log_bytes_written_total[5m])", "legendFormat": "bytes/s" }
+      ],
+      "fieldConfig": { "defaults": { "unit": "Bps" } }
+    },
+    {
+      "type": "stat",
+      "title": "Existing Log Files",
+      "id": 13,
+      "gridPos": { "x": 16, "y": 15, "w": 4, "h": 3 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [{ "refId": "A", "expr": "cart_event_log_files_existing" }],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
+    },
+    {
+      "type": "stat",
+      "title": "Last Append Age (s)",
+      "id": 14,
+      "gridPos": { "x": 20, "y": 15, "w": 4, "h": 3 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "(time() - cart_event_log_last_append_unix)" }
+      ],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
+    },
+    {
+      "type": "stat",
+      "title": "Replay Failures Total",
+      "id": 15,
+      "gridPos": { "x": 16, "y": 18, "w": 4, "h": 3 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [{ "refId": "A", "expr": "cart_event_log_replay_failures_total" }],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
+    },
+    {
+      "type": "stat",
+      "title": "Replay Duration p95 (5m)",
+      "id": 16,
+      "gridPos": { "x": 20, "y": 18, "w": 4, "h": 3 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        {
+          "refId": "A",
+            "expr": "histogram_quantile(0.95, sum(rate(cart_event_log_replay_duration_seconds_bucket[5m])) by (le))"
+        }
+      ],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "fieldConfig": { "defaults": { "unit": "s" } } }
+    },
+
+    {
+      "type": "row",
+      "title": "Grain Lifecycle",
+      "gridPos": { "x": 0, "y": 21, "w": 24, "h": 1 },
+      "id": 17,
+      "collapsed": false
+    },
+    {
+      "type": "timeseries",
+      "title": "Spawn & Lookup Rates (1m)",
+      "id": 18,
+      "gridPos": { "x": 0, "y": 22, "w": 12, "h": 8 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "rate(cart_grain_spawned_total[1m])", "legendFormat": "spawns/s" },
+        { "refId": "B", "expr": "rate(cart_grain_lookups_total[1m])", "legendFormat": "lookups/s" }
+      ]
+    },
+    {
+      "type": "stat",
+      "title": "Negotiations Rate (5m)",
+      "id": 19,
+      "gridPos": { "x": 12, "y": 22, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        { "refId": "A", "expr": "rate(cart_remote_negotiation_total[5m])" }
+      ],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "orientation": "horizontal" }
+    },
+    {
+      "type": "stat",
+      "title": "Mutations Total",
+      "id": 20,
+      "gridPos": { "x": 18, "y": 22, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [{ "refId": "A", "expr": "cart_mutations_total" }],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
+    },
+
+    {
+      "type": "row",
+      "title": "Event Log Errors",
+      "gridPos": { "x": 0, "y": 30, "w": 24, "h": 1 },
+      "id": 21,
+      "collapsed": false
+    },
+    {
+      "type": "stat",
+      "title": "Unknown Event Types",
+      "id": 22,
+      "gridPos": { "x": 0, "y": 31, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [{ "refId": "A", "expr": "cart_event_log_unknown_types_total" }],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
+    },
+    {
+      "type": "stat",
+      "title": "Event Mutation Errors",
+      "id": 23,
+      "gridPos": { "x": 6, "y": 31, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [{ "refId": "A", "expr": "cart_event_log_mutation_errors_total" }],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
+    },
+    {
+      "type": "stat",
+      "title": "Replay Success Total",
+      "id": 24,
+      "gridPos": { "x": 12, "y": 31, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [{ "refId": "A", "expr": "cart_event_log_replay_total" }],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
+    },
+    {
+      "type": "stat",
+      "title": "Replay Duration p50 (5m)",
+      "id": 25,
+      "gridPos": { "x": 18, "y": 31, "w": 6, "h": 4 },
+      "datasource": "${DS_PROMETHEUS}",
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "histogram_quantile(0.50, sum(rate(cart_event_log_replay_duration_seconds_bucket[5m])) by (le))"
+        }
+      ],
+      "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "fieldConfig": { "defaults": { "unit": "s" } } }
+    }
+  ],
+  "templating": {
+    "list": [
+      {
+        "name": "DS_PROMETHEUS",
+        "label": "Prometheus",
+        "type": "datasource",
+        "query": "prometheus",
+        "current": { "text": "Prometheus", "value": "Prometheus" }
+      }
+    ]
+  },
+  "time": {
+    "from": "now-6h",
+    "to": "now"
+  },
+  "timepicker": {
+    "refresh_intervals": ["5s","10s","30s","1m","5m","15m","30m","1h"],
+    "time_options": ["5m","15m","30m","1h","6h","12h","24h","2d","7d"]
+  }
+}

grain-pool.go

@@ -1,244 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"log"
-	"sync"
-	"time"
-
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-)
-
-// grain-pool.go
-//
-// Migration Note:
-//   This file has been migrated to use uint64 cart keys internally (derived
-//   from the new CartID base62 representation). For backward compatibility,
-//   a deprecated legacy map keyed by CartId is maintained so existing code
-//   that directly indexes pool.grains with a CartId continues to compile
-//   until the full refactor across SyncedPool / remoteIndex is completed.
-//
-//   Authoritative storage: grains (map[uint64]*CartGrain)
-//   Legacy compatibility:  grainsLegacy (map[CartId]*CartGrain) - kept in sync.
-//
-//   Once all external usages are updated to rely on helper accessors,
-//   grainsLegacy can be removed.
-//
-// ---------------------------------------------------------------------------
-
-var (
-	poolGrains = promauto.NewGauge(prometheus.GaugeOpts{
-		Name: "cart_grains_in_pool",
-		Help: "The total number of grains in the pool",
-	})
-	poolSize = promauto.NewGauge(prometheus.GaugeOpts{
-		Name: "cart_pool_size",
-		Help: "The total number of mutations",
-	})
-	poolUsage = promauto.NewGauge(prometheus.GaugeOpts{
-		Name: "cart_grain_pool_usage",
-		Help: "The current usage of the grain pool",
-	})
-)
-
-// GrainPool interface remains legacy-compatible.
-type GrainPool interface {
-	Apply(id CartId, mutation interface{}) (*CartGrain, error)
-	Get(id CartId) (*CartGrain, error)
-}
-
-// Ttl keeps expiry info
-type Ttl struct {
-	Expires time.Time
-	Grain   *CartGrain
-}
-
-// GrainLocalPool now stores grains keyed by uint64 (CartKey).
-type GrainLocalPool struct {
-	mu       sync.RWMutex
-	grains   map[uint64]*CartGrain // authoritative only
-	expiry   []Ttl
-	spawn    func(id CartId) (*CartGrain, error)
-	Ttl      time.Duration
-	PoolSize int
-}
-
-// NewGrainLocalPool constructs a new pool.
-func NewGrainLocalPool(size int, ttl time.Duration, spawn func(id CartId) (*CartGrain, error)) *GrainLocalPool {
-	ret := &GrainLocalPool{
-		spawn:    spawn,
-		grains:   make(map[uint64]*CartGrain),
-		expiry:   make([]Ttl, 0),
-		Ttl:      ttl,
-		PoolSize: size,
-	}
-	cartPurge := time.NewTicker(time.Minute)
-	go func() {
-		for range cartPurge.C {
-			ret.Purge()
-		}
-	}()
-	return ret
-}
-
-// keyFromCartId derives the uint64 key from a legacy CartId deterministically.
-func keyFromCartId(id CartId) uint64 {
-	return LegacyToCartKey(id)
-}
-
-// storeGrain indexes a grain in both maps.
-func (p *GrainLocalPool) storeGrain(id CartId, g *CartGrain) {
-	k := keyFromCartId(id)
-	p.grains[k] = g
-}
-
-// deleteGrain removes a grain from both maps.
-func (p *GrainLocalPool) deleteGrain(id CartId) {
-	k := keyFromCartId(id)
-	delete(p.grains, k)
-}
-
-// SetAvailable pre-populates placeholder entries (legacy signature).
-func (p *GrainLocalPool) SetAvailable(availableWithLastChangeUnix map[CartId]int64) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	for id := range availableWithLastChangeUnix {
-		k := keyFromCartId(id)
-		if _, ok := p.grains[k]; !ok {
-			p.grains[k] = nil
-			p.expiry = append(p.expiry, Ttl{
-				Expires: time.Now().Add(p.Ttl),
-				Grain:   nil,
-			})
-		}
-	}
-}
-
-// Purge removes expired grains.
-func (p *GrainLocalPool) Purge() {
-	lastChangeTime := time.Now().Add(-p.Ttl)
-	keepChanged := lastChangeTime.Unix()
-
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	for i := 0; i < len(p.expiry); i++ {
-		item := p.expiry[i]
-		if item.Grain == nil {
-			continue
-		}
-		if item.Expires.Before(time.Now()) {
-			if item.Grain.GetLastChange() > keepChanged {
-				log.Printf("Expired item %s changed, keeping", item.Grain.GetId())
-				if i < len(p.expiry)-1 {
-					p.expiry = append(p.expiry[:i], p.expiry[i+1:]...)
-					p.expiry = append(p.expiry, item)
-				} else {
-					// move last to end (noop)
-					p.expiry = append(p.expiry[:i], item)
-				}
-			} else {
-				log.Printf("Item %s expired", item.Grain.GetId())
-				p.deleteGrain(item.Grain.GetId())
-				if i < len(p.expiry)-1 {
-					p.expiry = append(p.expiry[:i], p.expiry[i+1:]...)
-				} else {
-					p.expiry = p.expiry[:i]
-				}
-			}
-		} else {
-			break
-		}
-	}
-}
-
-// GetGrains returns a legacy view of grains (copy) for compatibility.
-func (p *GrainLocalPool) GetGrains() map[CartId]*CartGrain {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	out := make(map[CartId]*CartGrain, len(p.grains))
-	for _, g := range p.grains {
-		if g != nil {
-			out[g.GetId()] = g
-		}
-	}
-	return out
-}
-
-// statsUpdate updates Prometheus gauges asynchronously.
-func (p *GrainLocalPool) statsUpdate() {
-	go func(size int) {
-		l := float64(size)
-		ps := float64(p.PoolSize)
-		poolUsage.Set(l / ps)
-		poolGrains.Set(l)
-		poolSize.Set(ps)
-	}(len(p.grains))
-}
-
-// GetGrain retrieves or spawns a grain (legacy id signature).
-func (p *GrainLocalPool) GetGrain(id CartId) (*CartGrain, error) {
-	grainLookups.Inc()
-	k := keyFromCartId(id)
-
-	p.mu.RLock()
-	grain, ok := p.grains[k]
-	p.mu.RUnlock()
-
-	var err error
-	if grain == nil || !ok {
-		p.mu.Lock()
-		// Re-check under write lock
-		grain, ok = p.grains[k]
-		if grain == nil || !ok {
-			// Capacity check
-			if len(p.grains) >= p.PoolSize && len(p.expiry) > 0 {
-				if p.expiry[0].Expires.Before(time.Now()) && p.expiry[0].Grain != nil {
-					oldId := p.expiry[0].Grain.GetId()
-					p.deleteGrain(oldId)
-					p.expiry = p.expiry[1:]
-				} else {
-					p.mu.Unlock()
-					return nil, fmt.Errorf("pool is full")
-				}
-			}
-			grain, err = p.spawn(id)
-			if err == nil {
-				p.storeGrain(id, grain)
-			}
-		}
-		p.mu.Unlock()
-		p.statsUpdate()
-	}
-
-	return grain, err
-}
-
-// Apply applies a mutation (legacy compatibility).
-func (p *GrainLocalPool) Apply(id CartId, mutation interface{}) (*CartGrain, error) {
-	grain, err := p.GetGrain(id)
-	if err != nil || grain == nil {
-		return nil, err
-	}
-	return grain.Apply(mutation, false)
-}
-
-// Get returns current state (legacy wrapper).
-func (p *GrainLocalPool) Get(id CartId) (*CartGrain, error) {
-	return p.GetGrain(id)
-}
-
-// DebugGrainCount returns counts for debugging.
-func (p *GrainLocalPool) DebugGrainCount() (authoritative int) {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	return len(p.grains)
-}
-
-// UnsafePointerToLegacyMap exposes the legacy map pointer (for transitional
-// tests that still poke the field directly). DO NOT rely on this long-term.
-func (p *GrainLocalPool) UnsafePointerToLegacyMap() uintptr {
-	// Legacy map removed; retained only to satisfy any transitional callers.
-	return 0
-}

grpc_integration_test.go

@@ -1,115 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"testing"
-	"time"
-
-	messages "git.tornberg.me/go-cart-actor/proto"
-	"google.golang.org/grpc"
-)
-
-// TestCartActorMutationAndState validates end-to-end gRPC mutation + state retrieval
-// against a locally started gRPC server (single-node scenario).
-// This test uses the new per-mutation AddItem RPC (breaking v2 API) to avoid external product fetch logic
-// fetching logic (FetchItem) which would require network I/O.
-func TestCartActorMutationAndState(t *testing.T) {
-	// Setup local grain pool + synced pool (no discovery, single host)
-	pool := NewGrainLocalPool(1024, time.Minute, spawn)
-	synced, err := NewSyncedPool(pool, "127.0.0.1", nil)
-	if err != nil {
-		t.Fatalf("NewSyncedPool error: %v", err)
-	}
-
-	// Start gRPC server (CartActor + ControlPlane) on :1337
-	grpcSrv, err := StartGRPCServer(":1337", pool, synced)
-	if err != nil {
-		t.Fatalf("StartGRPCServer error: %v", err)
-	}
-	defer grpcSrv.GracefulStop()
-
-	// Dial the local server
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-	defer cancel()
-	conn, err := grpc.DialContext(ctx, "127.0.0.1:1337",
-		grpc.WithInsecure(),
-		grpc.WithBlock(),
-	)
-	if err != nil {
-		t.Fatalf("grpc.Dial error: %v", err)
-	}
-	defer conn.Close()
-
-	cartClient := messages.NewCartActorClient(conn)
-
-	// Create a short cart id (<=16 chars so it fits into the fixed CartId 16-byte array cleanly)
-	cartID := fmt.Sprintf("cart-%d", time.Now().UnixNano())
-
-	// Build an AddItem payload (bypasses FetchItem to keep test deterministic)
-	addItem := &messages.AddItem{
-		ItemId:   1,
-		Quantity: 1,
-		Price:    1000,
-		OrgPrice: 1000,
-		Sku:      "test-sku",
-		Name:     "Test SKU",
-		Image:    "/img.png",
-		Stock:    2, // InStock
-		Tax:      2500,
-		Country:  "se",
-	}
-
-	// Issue AddItem RPC directly (breaking v2 API)
-	addResp, err := cartClient.AddItem(context.Background(), &messages.AddItemRequest{
-		CartId:          cartID,
-		ClientTimestamp: time.Now().Unix(),
-		Payload:         addItem,
-	})
-	if err != nil {
-		t.Fatalf("AddItem RPC error: %v", err)
-	}
-	if addResp.StatusCode != 200 {
-		t.Fatalf("AddItem returned non-200 status: %d, error: %s", addResp.StatusCode, addResp.GetError())
-	}
-
-	// Validate the response state (from AddItem)
-	state := addResp.GetState()
-	if state == nil {
-		t.Fatalf("AddItem response state is nil")
-	}
-
-	// (Removed obsolete Mutate response handling)
-
-	if len(state.Items) != 1 {
-		t.Fatalf("Expected 1 item after AddItem, got %d", len(state.Items))
-	}
-	if state.Items[0].Sku != "test-sku" {
-		t.Fatalf("Unexpected item SKU: %s", state.Items[0].Sku)
-	}
-
-	// Issue GetState RPC
-	getResp, err := cartClient.GetState(context.Background(), &messages.StateRequest{
-		CartId: cartID,
-	})
-	if err != nil {
-		t.Fatalf("GetState RPC error: %v", err)
-	}
-	if getResp.StatusCode != 200 {
-		t.Fatalf("GetState returned non-200 status: %d, error: %s", getResp.StatusCode, getResp.GetError())
-	}
-
-	state2 := getResp.GetState()
-	if state2 == nil {
-		t.Fatalf("GetState response state is nil")
-	}
-
-	if len(state2.Items) != 1 {
-		t.Fatalf("Expected 1 item in GetState, got %d", len(state2.Items))
-	}
-	if state2.Items[0].Sku != "test-sku" {
-		t.Fatalf("Unexpected SKU in GetState: %s", state2.Items[0].Sku)
-	}
-}
-
-// Legacy serialization helper removed (oneof envelope used directly)

grpc_server.go

@@ -1,280 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"net"
-	"time"
-
-	messages "git.tornberg.me/go-cart-actor/proto"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/reflection"
-)
-
-// cartActorGRPCServer implements the CartActor and ControlPlane gRPC services.
-// It delegates cart operations to a grain pool and cluster operations to a synced pool.
-type cartActorGRPCServer struct {
-	messages.UnimplementedCartActorServer
-	messages.UnimplementedControlPlaneServer
-
-	pool       GrainPool   // For cart state mutations and queries
-	syncedPool *SyncedPool // For cluster membership and control
-}
-
-// NewCartActorGRPCServer creates and initializes the server.
-func NewCartActorGRPCServer(pool GrainPool, syncedPool *SyncedPool) *cartActorGRPCServer {
-	return &cartActorGRPCServer{
-		pool:       pool,
-		syncedPool: syncedPool,
-	}
-}
-
-// applyMutation routes a single cart mutation to the target grain (used by per-mutation RPC handlers).
-func (s *cartActorGRPCServer) applyMutation(cartID string, mutation interface{}) *messages.CartMutationReply {
-	// Canonicalize or preserve legacy id (do NOT hash-rewrite legacy textual ids)
-	cid, _, wasBase62, cerr := CanonicalizeOrLegacy(cartID)
-	if cerr != nil {
-		return &messages.CartMutationReply{
-			StatusCode:      500,
-			Result:          &messages.CartMutationReply_Error{Error: fmt.Sprintf("cart_id canonicalization failed: %v", cerr)},
-			ServerTimestamp: time.Now().Unix(),
-		}
-	}
-	_ = wasBase62 // placeholder; future: propagate canonical id in reply metadata
-	legacy := CartIDToLegacy(cid)
-	grain, err := s.pool.Apply(legacy, mutation)
-	if err != nil {
-		return &messages.CartMutationReply{
-			StatusCode:      500,
-			Result:          &messages.CartMutationReply_Error{Error: err.Error()},
-			ServerTimestamp: time.Now().Unix(),
-		}
-	}
-	cartState := ToCartState(grain)
-	return &messages.CartMutationReply{
-		StatusCode:      200,
-		Result:          &messages.CartMutationReply_State{State: cartState},
-		ServerTimestamp: time.Now().Unix(),
-	}
-}
-
-func (s *cartActorGRPCServer) AddRequest(ctx context.Context, req *messages.AddRequestRequest) (*messages.CartMutationReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.CartMutationReply{
-			StatusCode:      400,
-			Result:          &messages.CartMutationReply_Error{Error: "cart_id is required"},
-			ServerTimestamp: time.Now().Unix(),
-		}, nil
-	}
-	return s.applyMutation(req.GetCartId(), req.GetPayload()), nil
-}
-
-func (s *cartActorGRPCServer) AddItem(ctx context.Context, req *messages.AddItemRequest) (*messages.CartMutationReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.CartMutationReply{
-			StatusCode:      400,
-			Result:          &messages.CartMutationReply_Error{Error: "cart_id is required"},
-			ServerTimestamp: time.Now().Unix(),
-		}, nil
-	}
-	return s.applyMutation(req.GetCartId(), req.GetPayload()), nil
-}
-
-func (s *cartActorGRPCServer) RemoveItem(ctx context.Context, req *messages.RemoveItemRequest) (*messages.CartMutationReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.CartMutationReply{
-			StatusCode:      400,
-			Result:          &messages.CartMutationReply_Error{Error: "cart_id is required"},
-			ServerTimestamp: time.Now().Unix(),
-		}, nil
-	}
-	return s.applyMutation(req.GetCartId(), req.GetPayload()), nil
-}
-
-func (s *cartActorGRPCServer) RemoveDelivery(ctx context.Context, req *messages.RemoveDeliveryRequest) (*messages.CartMutationReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.CartMutationReply{
-			StatusCode:      400,
-			Result:          &messages.CartMutationReply_Error{Error: "cart_id is required"},
-			ServerTimestamp: time.Now().Unix(),
-		}, nil
-	}
-	return s.applyMutation(req.GetCartId(), req.GetPayload()), nil
-}
-
-func (s *cartActorGRPCServer) ChangeQuantity(ctx context.Context, req *messages.ChangeQuantityRequest) (*messages.CartMutationReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.CartMutationReply{
-			StatusCode:      400,
-			Result:          &messages.CartMutationReply_Error{Error: "cart_id is required"},
-			ServerTimestamp: time.Now().Unix(),
-		}, nil
-	}
-	return s.applyMutation(req.GetCartId(), req.GetPayload()), nil
-}
-
-func (s *cartActorGRPCServer) SetDelivery(ctx context.Context, req *messages.SetDeliveryRequest) (*messages.CartMutationReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.CartMutationReply{
-			StatusCode:      400,
-			Result:          &messages.CartMutationReply_Error{Error: "cart_id is required"},
-			ServerTimestamp: time.Now().Unix(),
-		}, nil
-	}
-	return s.applyMutation(req.GetCartId(), req.GetPayload()), nil
-}
-
-func (s *cartActorGRPCServer) SetPickupPoint(ctx context.Context, req *messages.SetPickupPointRequest) (*messages.CartMutationReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.CartMutationReply{
-			StatusCode:      400,
-			Result:          &messages.CartMutationReply_Error{Error: "cart_id is required"},
-			ServerTimestamp: time.Now().Unix(),
-		}, nil
-	}
-	return s.applyMutation(req.GetCartId(), req.GetPayload()), nil
-}
-
-/*
-Checkout RPC removed. Checkout is handled at the HTTP layer (PoolServer.HandleCheckout).
-*/
-
-func (s *cartActorGRPCServer) SetCartItems(ctx context.Context, req *messages.SetCartItemsRequest) (*messages.CartMutationReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.CartMutationReply{
-			StatusCode:      400,
-			Result:          &messages.CartMutationReply_Error{Error: "cart_id is required"},
-			ServerTimestamp: time.Now().Unix(),
-		}, nil
-	}
-	return s.applyMutation(req.GetCartId(), req.GetPayload()), nil
-}
-
-func (s *cartActorGRPCServer) OrderCompleted(ctx context.Context, req *messages.OrderCompletedRequest) (*messages.CartMutationReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.CartMutationReply{
-			StatusCode:      400,
-			Result:          &messages.CartMutationReply_Error{Error: "cart_id is required"},
-			ServerTimestamp: time.Now().Unix(),
-		}, nil
-	}
-	return s.applyMutation(req.GetCartId(), req.GetPayload()), nil
-}
-
-// GetState retrieves the current state of a cart grain.
-func (s *cartActorGRPCServer) GetState(ctx context.Context, req *messages.StateRequest) (*messages.StateReply, error) {
-	if req.GetCartId() == "" {
-		return &messages.StateReply{
-			StatusCode: 400,
-			Result:     &messages.StateReply_Error{Error: "cart_id is required"},
-		}, nil
-	}
-	// Canonicalize / upgrade incoming cart id (preserve legacy strings)
-	cid, _, _, cerr := CanonicalizeOrLegacy(req.GetCartId())
-	if cerr != nil {
-		return &messages.StateReply{
-			StatusCode: 500,
-			Result:     &messages.StateReply_Error{Error: fmt.Sprintf("cart_id canonicalization failed: %v", cerr)},
-		}, nil
-	}
-	legacy := CartIDToLegacy(cid)
-
-	grain, err := s.pool.Get(legacy)
-	if err != nil {
-		return &messages.StateReply{
-			StatusCode: 500,
-			Result:     &messages.StateReply_Error{Error: err.Error()},
-		}, nil
-	}
-
-	cartState := ToCartState(grain)
-
-	return &messages.StateReply{
-		StatusCode: 200,
-		Result:     &messages.StateReply_State{State: cartState},
-	}, nil
-}
-
-// ControlPlane: Ping
-func (s *cartActorGRPCServer) Ping(ctx context.Context, _ *messages.Empty) (*messages.PingReply, error) {
-	return &messages.PingReply{
-		Host:     s.syncedPool.Hostname,
-		UnixTime: time.Now().Unix(),
-	}, nil
-}
-
-// ControlPlane: Negotiate (merge host views)
-func (s *cartActorGRPCServer) Negotiate(ctx context.Context, req *messages.NegotiateRequest) (*messages.NegotiateReply, error) {
-	hostSet := make(map[string]struct{})
-	// Caller view
-	for _, h := range req.GetKnownHosts() {
-		if h != "" {
-			hostSet[h] = struct{}{}
-		}
-	}
-	// This host
-	hostSet[s.syncedPool.Hostname] = struct{}{}
-	// Known remotes
-	s.syncedPool.mu.RLock()
-	for h := range s.syncedPool.remoteHosts {
-		hostSet[h] = struct{}{}
-	}
-	s.syncedPool.mu.RUnlock()
-
-	out := make([]string, 0, len(hostSet))
-	for h := range hostSet {
-		out = append(out, h)
-	}
-	return &messages.NegotiateReply{Hosts: out}, nil
-}
-
-// ControlPlane: GetCartIds (locally owned carts only)
-func (s *cartActorGRPCServer) GetCartIds(ctx context.Context, _ *messages.Empty) (*messages.CartIdsReply, error) {
-	s.syncedPool.local.mu.RLock()
-	ids := make([]string, 0, len(s.syncedPool.local.grains))
-	for _, g := range s.syncedPool.local.grains {
-		if g == nil {
-			continue
-		}
-		ids = append(ids, g.GetId().String())
-	}
-	s.syncedPool.local.mu.RUnlock()
-	return &messages.CartIdsReply{CartIds: ids}, nil
-}
-
-// ControlPlane: Closing (peer shutdown notification)
-func (s *cartActorGRPCServer) Closing(ctx context.Context, req *messages.ClosingNotice) (*messages.OwnerChangeAck, error) {
-	if req.GetHost() != "" {
-		s.syncedPool.RemoveHost(req.GetHost())
-	}
-	return &messages.OwnerChangeAck{
-		Accepted: true,
-		Message:  "removed host",
-	}, nil
-}
-
-// StartGRPCServer configures and starts the unified gRPC server on the given address.
-// It registers both the CartActor and ControlPlane services.
-func StartGRPCServer(addr string, pool GrainPool, syncedPool *SyncedPool) (*grpc.Server, error) {
-	lis, err := net.Listen("tcp", addr)
-	if err != nil {
-		return nil, fmt.Errorf("failed to listen: %w", err)
-	}
-
-	grpcServer := grpc.NewServer()
-	server := NewCartActorGRPCServer(pool, syncedPool)
-
-	messages.RegisterCartActorServer(grpcServer, server)
-	messages.RegisterControlPlaneServer(grpcServer, server)
-	reflection.Register(grpcServer)
-
-	log.Printf("gRPC server listening on %s", addr)
-	go func() {
-		if err := grpcServer.Serve(lis); err != nil {
-			log.Fatalf("failed to serve gRPC: %v", err)
-		}
-	}()
-
-	return grpcServer, nil
-}

k6/README.md

@@ -0,0 +1,174 @@
+# k6 Load Tests for Cart API
+
+This directory contains a k6 script (`cart_load_test.js`) to stress and observe the cart actor HTTP API.
+
+## Contents
+
+- `cart_load_test.js` – primary k6 scenario script
+- `README.md` – this file
+
+## Prerequisites
+
+- Node not required (k6 runs standalone)
+- k6 installed (>= v0.43 recommended)
+- Prometheus + Grafana (optional) if you want to correlate with the dashboard you generated
+- A running cart service exposing HTTP endpoints at (default) `http://localhost:8080/cart`
+
+## Endpoints Exercised
+
+The script exercises (per iteration):
+
+1. `GET /cart/` – ensure / fetch cart state (creates cart if missing; sets `cartid` & `cartowner` cookies)
+2. `POST /cart/` – add item mutation (random SKU & quantity)
+3. `GET /cart/` – fetch after mutations
+4. `GET /cart/checkout` – occasionally (~2% of iterations) to simulate checkout start
+
+You can extend it easily to hit deliveries, quantity changes, or removal endpoints.
+
+## Environment Variables
+
+| Variable        | Purpose                                      | Default                 |
+|-----------------|----------------------------------------------|-------------------------|
+| `BASE_URL`      | Base URL root (either host or host/cart)     | `http://localhost:8080/cart` |
+| `VUS`           | VUs for steady_mutations scenario            | `20`                    |
+| `DURATION`      | Duration for steady_mutations scenario       | `5m`                    |
+| `RAMP_TARGET`   | Peak VUs for ramp_up scenario                 | `50`                    |
+
+You can also disable one scenario by editing `options.scenarios` inside the script.
+
+Example run:
+
+```bash
+k6 run \
+  -e BASE_URL=https://cart.prod.example.com/cart \
+  -e VUS=40 \
+  -e DURATION=10m \
+  -e RAMP_TARGET=120 \
+  k6/cart_load_test.js
+```
+
+## Metrics (Custom)
+
+The script defines additional k6 metrics:
+
+- `cart_add_item_duration` (Trend) – latency of POST add item
+- `cart_fetch_duration` (Trend) – latency of GET cart state
+- `cart_checkout_duration` (Trend) – latency of checkout
+- `cart_items_added` (Counter) – successful add item operations
+- `cart_checkout_calls` (Counter) – successful checkout calls
+
+Thresholds (in `options.thresholds`) enforce basic SLO:
+- Mutation failure rate < 2%
+- p90 mutation latency < 800 ms
+- p99 overall HTTP latency < 1500 ms
+
+Adjust thresholds to your environment if they trigger prematurely.
+
+## Cookies & Stickiness
+
+The script preserves:
+- `cartid` – cart identity (server sets expiry separately)
+- `cartowner` – owning host for sticky routing
+
+If your load balancer or ingress enforces affinity based on these cookies, traffic will naturally concentrate on the originally claimed host for each cart under test.
+
+## SKU Set
+
+SKUs used (randomly selected each mutation):
+
+```
+778290 778345 778317 778277 778267 778376 778244 778384
+778365 778377 778255 778286 778246 778270 778266 778285
+778329 778425 778407 778418 778430 778469 778358 778351
+778319 778307 778278 778251 778253 778261 778263 778273
+778281 778294 778297 778302
+```
+
+To add/remove SKUs, edit the `SKUS` array. Keeping it non-empty and moderately sized helps randomization.
+
+## Extending the Script
+
+### Add Quantity Change
+
+```js
+function changeQuantity(itemId, newQty) {
+  const payload = JSON.stringify({ Id: itemId, Qty: newQty });
+  http.put(baseUrl() + '/', payload, { headers: headers() });
+}
+```
+
+### Remove Item
+
+```js
+function removeItem(itemId) {
+  http.del(baseUrl() + '/' + itemId, null, { headers: headers() });
+}
+```
+
+### Add Delivery
+
+```js
+function addDelivery(itemIds) {
+  const payload = JSON.stringify({ provider: "POSTNORD", items: itemIds });
+  http.post(baseUrl() + '/delivery', payload, { headers: headers() });
+}
+```
+
+You can integrate these into the iteration loop with probabilities.
+
+## Output Summary
+
+`handleSummary` outputs a JSON summary to stdout:
+- Average & p95 mutation latencies (if present)
+- Fetch p95
+- Checkout count
+- Check statuses
+
+Redirect or parse that output for CI pipelines.
+
+## Running in CI
+
+Use shorter durations (e.g. `DURATION=2m VUS=10`) to keep builds fast. Fail build on threshold breaches:
+
+```bash
+k6 run -e BASE_URL=$TARGET -e VUS=10 -e DURATION=2m k6/cart_load_test.js || exit 1
+```
+
+## Correlating with Prometheus / Grafana
+
+During load, observe:
+- `cart_mutations_total` growth and latency histograms
+- Event log write rate (`cart_event_log_appends_total`)
+- Pool usage (`cart_grain_pool_usage`) and spawn rate (`cart_grain_spawned_total`)
+- Failure counters (`cart_mutation_failures_total`) ensure they remain low
+
+If mutation latency spikes without high error rate, inspect external dependencies (e.g., product fetcher or Klarna endpoints).
+
+## Common Tuning Tips
+
+| Symptom                             | Potential Adjustment                              |
+|------------------------------------|---------------------------------------------------|
+| High latency p99                   | Increase CPU/memory, optimize mutation handlers   |
+| Pool at capacity                   | Raise pool size argument or TTL                   |
+| Frequent cart eviction mid-test    | Confirm TTL is sliding (now 2h on mutation)       |
+| High replay duration               | Consider snapshot + truncate event logs           |
+| Uneven host load                   | Verify `cartowner` cookie is respected upstream   |
+
+## Safety / Load Guardrails
+
+- Start with low VUs (5–10) and short duration.
+- Scale incrementally to find saturation points.
+- If using production endpoints, coordinate off-peak runs.
+
+## License / Attribution
+
+This test script is tailored for your internal cart actor system; adapt freely. k6 is open-source (AGPL v3). Ensure compliance if redistributing.
+
+---
+
+Feel free to request:
+- A variant script for spike tests
+- WebSocket / long poll integration (if added later)
+- Synthetic error injection harness
+
+Happy load testing!

k6/cart_load_test.js

@@ -0,0 +1,248 @@
+import http from "k6/http";
+import { check, sleep, group } from "k6";
+import { Counter, Trend } from "k6/metrics";
+
+// ---------------- Configuration ----------------
+export const options = {
+    // Adjust vus/duration for your environment
+    scenarios: {
+        steady_mutations: {
+            executor: "constant-vus",
+            vus: __ENV.VUS ? parseInt(__ENV.VUS, 10) : 20,
+            duration: __ENV.DURATION || "5m",
+            gracefulStop: "30s",
+        },
+        ramp_up: {
+            executor: "ramping-vus",
+            startVUs: 0,
+            stages: [
+                {
+                    duration: "1m",
+                    target: __ENV.RAMP_TARGET
+                        ? parseInt(__ENV.RAMP_TARGET, 10)
+                        : 50,
+                },
+                {
+                    duration: "1m",
+                    target: __ENV.RAMP_TARGET
+                        ? parseInt(__ENV.RAMP_TARGET, 10)
+                        : 50,
+                },
+                { duration: "1m", target: 0 },
+            ],
+            gracefulStop: "30s",
+            startTime: "30s",
+        },
+    },
+    thresholds: {
+        http_req_failed: ["rate<0.02"], // < 2% failures
+        http_req_duration: ["p(90)<800", "p(99)<1500"], // latency SLO
+        "cart_add_item_duration{op:add}": ["p(90)<800"],
+        "cart_fetch_duration{op:get}": ["p(90)<600"],
+    },
+    summaryTrendStats: ["avg", "min", "med", "max", "p(90)", "p(95)", "p(99)"],
+};
+
+// ---------------- Metrics ----------------
+const addItemTrend = new Trend("cart_add_item_duration", true);
+const fetchTrend = new Trend("cart_fetch_duration", true);
+const checkoutTrend = new Trend("cart_checkout_duration", true);
+const addedItemsCounter = new Counter("cart_items_added");
+const checkoutCounter = new Counter("cart_checkout_calls");
+
+// ---------------- SKUs ----------------
+const SKUS = [
+    "778290",
+    "778345",
+    "778317",
+    "778277",
+    "778267",
+    "778376",
+    "778244",
+    "778384",
+    "778365",
+    "778377",
+    "778255",
+    "778286",
+    "778246",
+    "778270",
+    "778266",
+    "778285",
+    "778329",
+    "778425",
+    "778407",
+    "778418",
+    "778430",
+    "778469",
+    "778358",
+    "778351",
+    "778319",
+    "778307",
+    "778278",
+    "778251",
+    "778253",
+    "778261",
+    "778263",
+    "778273",
+    "778281",
+    "778294",
+    "778297",
+    "778302",
+];
+
+// ---------------- Helpers ----------------
+function randomSku() {
+    return SKUS[Math.floor(Math.random() * SKUS.length)];
+}
+function randomQty() {
+    return 1 + Math.floor(Math.random() * 3); // 1..3
+}
+function baseUrl() {
+    const u = __ENV.BASE_URL || "http://localhost:8080/cart";
+    // Allow user to pass either root host or full /cart path
+    return u.endsWith("/cart") ? u : u.replace(/\/+$/, "") + "/cart";
+}
+function extractCookie(res, name) {
+    const cookies = res.cookies[name];
+    if (!cookies || cookies.length === 0) return null;
+    return cookies[0].value;
+}
+function withCookies(headers, cookieJar) {
+    if (!cookieJar || Object.keys(cookieJar).length === 0) return headers;
+    const cookieStr = Object.entries(cookieJar)
+        .map(([k, v]) => `${k}=${v}`)
+        .join("; ");
+    return { ...headers, Cookie: cookieStr };
+}
+
+// Maintain cart + owner cookies per VU
+let cartState = {
+    cartid: null,
+    cartowner: null,
+};
+
+// Refresh cookies from response
+function updateCookies(res) {
+    const cid = extractCookie(res, "cartid");
+    if (cid) cartState.cartid = cid;
+    const owner = extractCookie(res, "cartowner");
+    if (owner) cartState.cartowner = owner;
+}
+
+// Build headers
+function headers() {
+    const h = { "Content-Type": "application/json" };
+    const jar = {};
+    if (cartState.cartid) jar["cartid"] = cartState.cartid;
+    if (cartState.cartowner) jar["cartowner"] = cartState.cartowner;
+    return withCookies(h, jar);
+}
+
+// Ensure cart exists (GET /)
+function ensureCart() {
+    if (cartState.cartid) return;
+    const res = http.get(baseUrl() + "/", { headers: headers() });
+    updateCookies(res);
+    check(res, {
+        "ensure cart status 200": (r) => r.status === 200,
+        "ensure cart has id": () => !!cartState.cartid,
+    });
+}
+
+// Add random item
+function addRandomItem() {
+    const payload = JSON.stringify({
+        sku: randomSku(),
+        quantity: randomQty(),
+        country: "no",
+    });
+    const start = Date.now();
+    const res = http.post(baseUrl(), payload, { headers: headers() });
+    const dur = Date.now() - start;
+    addItemTrend.add(dur, { op: "add" });
+    if (res.status === 200) {
+        addedItemsCounter.add(1);
+    }
+    updateCookies(res);
+    check(res, {
+        "add item status ok": (r) => r.status === 200,
+    });
+}
+
+// Fetch cart state
+function fetchCart() {
+    const start = Date.now();
+    const res = http.get(baseUrl() + "/", { headers: headers() });
+    const dur = Date.now() - start;
+    fetchTrend.add(dur, { op: "get" });
+    updateCookies(res);
+    check(res, { "fetch status ok": (r) => r.status === 200 });
+}
+
+// Occasional checkout trigger
+function maybeCheckout() {
+    if (!cartState.cartid) return;
+    // // Small probability
+    // if (Math.random() < 0.02) {
+    //     const start = Date.now();
+    //     const res = http.get(baseUrl() + "/checkout", { headers: headers() });
+    //     const dur = Date.now() - start;
+    //     checkoutTrend.add(dur, { op: "checkout" });
+    //     updateCookies(res);
+    //     if (res.status === 200) checkoutCounter.add(1);
+    //     check(res, { "checkout status ok": (r) => r.status === 200 });
+    // }
+}
+
+// ---------------- k6 lifecycle ----------------
+export function setup() {
+    // Provide SKU list length for summary
+    return { skuCount: SKUS.length };
+}
+
+export default function (data) {
+    group("cart flow", () => {
+        // Create or reuse cart
+        ensureCart();
+
+        // Random number of item mutations per iteration (1..5)
+        const ops = 1 + Math.floor(Math.random() * 5);
+        for (let i = 0; i < ops; i++) {
+            addRandomItem();
+        }
+
+        // Fetch state
+        fetchCart();
+
+        // Optional checkout attempt
+        maybeCheckout();
+    });
+
+    // Small think time
+    sleep(Math.random() * 0.5);
+}
+
+export function teardown(data) {
+    // Optionally we could GET confirmation or clear cart cookie
+    // Not implemented for load purpose.
+    console.log(`Test complete. SKU count: ${data.skuCount}`);
+}
+
+// ---------------- Summary ----------------
+export function handleSummary(data) {
+    return {
+        stdout: JSON.stringify(
+            {
+                metrics: {
+                    mutations_avg: data.metrics.cart_add_item_duration?.avg,
+                    mutations_p95: data.metrics.cart_add_item_duration?.p(95),
+                    fetch_p95: data.metrics.cart_fetch_duration?.p(95),
+                    checkout_count: data.metrics.cart_checkout_calls?.count,
+                },
+                checks: data.root_checks,
+            },
+            null,
+            2,
+        ),
+    };
+}

main.go

@@ -12,7 +12,9 @@ import (
 	"syscall"
 	"time"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	"git.tornberg.me/go-cart-actor/pkg/actor"
+	"git.tornberg.me/go-cart-actor/pkg/discovery"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -43,10 +45,16 @@ func spawn(id CartId) (*CartGrain, error) {
 		Deliveries:     []*CartDelivery{},
 		Id:             id,
 		Items:          []*CartItem{},
-		// storageMessages removed (legacy event log deprecated)
 		TotalPrice:     0,
 	}
-	err := loadMessages(ret, id)
+	// Set baseline lastChange at spawn; replay may update it to last event timestamp.
+	ret.lastChange = time.Now()
+	ret.lastAccess = time.Now()
+
+	// Legacy loadMessages (no-op) retained; then replay append-only event log
+	//_ = loadMessages(ret, id)
+	err := ReplayCartEvents(ret, id)
+
 	return ret, err
 }
 
@@ -55,42 +63,25 @@ func init() {
 }
 
 type App struct {
-	pool    *GrainLocalPool
+	pool    *CartPool
 	storage *DiskStorage
 }
 
-func (a *App) Save() error {
-	hasChanges := false
-	a.pool.mu.RLock()
-	defer a.pool.mu.RUnlock()
-	for id, grain := range a.pool.GetGrains() {
-		if grain == nil {
-			continue
-		}
-		if grain.GetLastChange() > a.storage.LastSaves[id] {
-			hasChanges = true
-			err := a.storage.Store(id, grain)
-			if err != nil {
-				log.Printf("Error saving grain %s: %v\n", id, err)
-			}
-		}
-	}
+// func (a *App) Save() error {
+// 	for id, grain := range a.pool.SnapshotGrains() {
+// 		if grain == nil {
+// 			continue
+// 		}
+// 		if grain.GetLastChange().After(a.storage.LastSaves[uint64(id)]) {
 
-	if !hasChanges {
-		return nil
-	}
-	return a.storage.saveState()
-}
-
-func (a *App) HandleSave(w http.ResponseWriter, r *http.Request) {
-	err := a.Save()
-	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		w.Write([]byte(err.Error()))
-	} else {
-		w.WriteHeader(http.StatusCreated)
-	}
-}
+// 			err := a.storage.Store(id, grain)
+// 			if err != nil {
+// 				log.Printf("Error saving grain %s: %v\n", id, err)
+// 			}
+// 		}
+// 	}
+// 	return nil
+// }
 
 var podIp = os.Getenv("POD_IP")
 var name = os.Getenv("POD_NAME")
@@ -118,25 +109,7 @@ func getCountryFromHost(host string) string {
 	return "se"
 }
 
-func getCheckoutOrder(host string, cartId CartId) *messages.CreateCheckoutOrder {
-	baseUrl := fmt.Sprintf("https://%s", host)
-	cartBaseUrl := os.Getenv("CART_BASE_URL")
-	if cartBaseUrl == "" {
-		cartBaseUrl = "https://cart.tornberg.me"
-	}
-	country := getCountryFromHost(host)
-
-	return &messages.CreateCheckoutOrder{
-		Terms:        fmt.Sprintf("%s/terms", baseUrl),
-		Checkout:     fmt.Sprintf("%s/checkout?order_id={checkout.order.id}", baseUrl),
-		Confirmation: fmt.Sprintf("%s/confirmation/{checkout.order.id}", baseUrl),
-		Validation:   fmt.Sprintf("%s/validation", cartBaseUrl),
-		Push:         fmt.Sprintf("%s/push?order_id={checkout.order.id}", cartBaseUrl),
-		Country:      country,
-	}
-}
-
-func GetDiscovery() Discovery {
+func GetDiscovery() discovery.Discovery {
 	if podIp == "" {
 		return nil
 	}
@@ -150,46 +123,42 @@ func GetDiscovery() Discovery {
 	if err != nil {
 		log.Fatalf("Error creating client: %v\n", err)
 	}
-	return NewK8sDiscovery(client)
+	return discovery.NewK8sDiscovery(client)
 }
 
 func main() {
-
-	storage, err := NewDiskStorage(fmt.Sprintf("data/%s_state.gob", name))
+	storage, err := NewDiskStorage(fmt.Sprintf("data/s_%s.gob", name))
 	if err != nil {
 		log.Printf("Error loading state: %v\n", err)
 	}
+	pool, err := NewCartPool(2*65535, 15*time.Minute, podIp, spawn, GetDiscovery())
+	if err != nil {
+		log.Fatalf("Error creating cart pool: %v\n", err)
+	}
 	app := &App{
-		pool:    NewGrainLocalPool(65535, 5*time.Minute, spawn),
+		pool:    pool,
 		storage: storage,
 	}
 
-	syncedPool, err := NewSyncedPool(app.pool, podIp, GetDiscovery())
+	grpcSrv, err := actor.NewControlServer[*CartGrain](":1337", pool)
 	if err != nil {
-		log.Fatalf("Error creating synced pool: %v\n", err)
-	}
-
-	// Start unified gRPC server (CartActor + ControlPlane) replacing legacy RPC server on :1337
-	// TODO: Remove any remaining legacy RPC server references and deprecated frame-based code after full gRPC migration is validated.
-	grpcSrv, err := StartGRPCServer(":1337", app.pool, syncedPool)
-	if err != nil {
-		log.Fatalf("Error starting gRPC server: %v\n", err)
+		log.Fatalf("Error starting control plane gRPC server: %v\n", err)
 	}
 	defer grpcSrv.GracefulStop()
 
-	go func() {
-		for range time.Tick(time.Minute * 10) {
-			err := app.Save()
-			if err != nil {
-				log.Printf("Error saving: %v\n", err)
-			}
-		}
-	}()
+	// go func() {
+	// 	for range time.Tick(time.Minute * 5) {
+	// 		err := app.Save()
+	// 		if err != nil {
+	// 			log.Printf("Error saving: %v\n", err)
+	// 		}
+	// 	}
+	// }()
 	orderHandler := &AmqpOrderHandler{
 		Url: amqpUrl,
 	}
 
-	syncedServer := NewPoolServer(syncedPool, fmt.Sprintf("%s, %s", name, podIp))
+	syncedServer := NewPoolServer(pool, fmt.Sprintf("%s, %s", name, podIp))
 	mux := http.NewServeMux()
 	mux.Handle("/cart/", http.StripPrefix("/cart", syncedServer.Serve()))
 	// only for local
@@ -206,16 +175,13 @@ func main() {
 	mux.Handle("/metrics", promhttp.Handler())
 	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
 		// Grain pool health: simple capacity check (mirrors previous GrainHandler.IsHealthy)
-		app.pool.mu.RLock()
-		grainCount := len(app.pool.grains)
-		capacity := app.pool.PoolSize
-		app.pool.mu.RUnlock()
+		grainCount, capacity := app.pool.LocalUsage()
 		if grainCount >= capacity {
 			w.WriteHeader(http.StatusInternalServerError)
 			w.Write([]byte("grain pool at capacity"))
 			return
 		}
-		if !syncedPool.IsHealthy() {
+		if !pool.IsHealthy() {
 			w.WriteHeader(http.StatusInternalServerError)
 			w.Write([]byte("control plane not healthy"))
 			return
@@ -248,7 +214,13 @@ func main() {
 				w.Write([]byte("no cart id to checkout is empty"))
 				return
 			}
-			cartId := ToCartId(cookie.Value)
+			parsed, ok := ParseCartId(cookie.Value)
+			if !ok {
+				w.WriteHeader(http.StatusBadRequest)
+				w.Write([]byte("invalid cart id format"))
+				return
+			}
+			cartId := parsed
 			order, err = syncedServer.CreateOrUpdateCheckout(r.Host, cartId)
 			if err != nil {
 				w.WriteHeader(http.StatusInternalServerError)
@@ -267,7 +239,7 @@ func main() {
 		w.Header().Set("Content-Type", "text/html; charset=utf-8")
 		w.Header().Set("Permissions-Policy", "payment=(self \"https://js.stripe.com\" \"https://m.stripe.network\" \"https://js.playground.kustom.co\")")
 		w.WriteHeader(http.StatusOK)
-		w.Write([]byte(fmt.Sprintf(tpl, order.HTMLSnippet)))
+		fmt.Fprintf(w, tpl, order.HTMLSnippet)
 	})
 	mux.HandleFunc("/confirmation/{order_id}", func(w http.ResponseWriter, r *http.Request) {
 
@@ -294,7 +266,7 @@ func main() {
 		}
 
 		w.WriteHeader(http.StatusOK)
-		w.Write([]byte(fmt.Sprintf(tpl, order.HTMLSnippet)))
+		fmt.Fprintf(w, tpl, order.HTMLSnippet)
 	})
 	mux.HandleFunc("/validate", func(w http.ResponseWriter, r *http.Request) {
 		log.Printf("Klarna order validation, method: %s", r.Method)
@@ -372,8 +344,9 @@ func main() {
 	go func() {
 		sig := <-sigs
 		fmt.Println("Shutting down due to signal:", sig)
-		go syncedPool.Close()
-		app.Save()
+		//app.Save()
+		pool.Close()
+
 		done <- true
 	}()
 
@@ -384,11 +357,19 @@ func main() {
 }
 
 func triggerOrderCompleted(err error, syncedServer *PoolServer, order *CheckoutOrder) error {
-	_, err = syncedServer.pool.Apply(ToCartId(order.MerchantReference1), &messages.OrderCreated{
+	mutation := &messages.OrderCreated{
 		OrderId: order.ID,
 		Status:  order.Status,
-	})
-	return err
+	}
+	cid, ok := ParseCartId(order.MerchantReference1)
+	if !ok {
+		return fmt.Errorf("invalid cart id in order reference: %s", order.MerchantReference1)
+	}
+	_, applyErr := syncedServer.pool.Apply(uint64(cid), mutation)
+	if applyErr == nil {
+		_ = AppendCartEvent(cid, mutation)
+	}
+	return applyErr
 }
 
 func confirmOrder(order *CheckoutOrder, orderHandler *AmqpOrderHandler) error {

multi_node_ownership_test.go

@@ -1,182 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"testing"
-	"time"
-
-	messages "git.tornberg.me/go-cart-actor/proto"
-	"google.golang.org/grpc"
-)
-
-// TestMultiNodeOwnershipNegotiation spins up two gRPC servers (nodeA, nodeB),
-// manually links their SyncedPools (bypassing AddRemote's fixed port assumption),
-// and verifies that only one node becomes the owner of a new cart while the
-// other can still apply a mutation via the remote proxy path.
-//
-// NOTE:
-//   - We manually inject RemoteHostGRPC entries because AddRemote() hard-codes
-//     port 1337; to run two distinct servers concurrently we need distinct ports.
-//   - This test asserts single ownership consistency rather than the complete
-//     quorum semantics (which depend on real discovery + AddRemote).
-func TestMultiNodeOwnershipNegotiation(t *testing.T) {
-	// Allocate distinct ports for the two nodes.
-	const (
-		addrA = "127.0.0.1:18081"
-		addrB = "127.0.0.1:18082"
-		hostA = "nodeA"
-		hostB = "nodeB"
-	)
-
-	// Create local grain pools.
-	poolA := NewGrainLocalPool(1024, time.Minute, spawn)
-	poolB := NewGrainLocalPool(1024, time.Minute, spawn)
-
-	// Create synced pools (no discovery).
-	syncedA, err := NewSyncedPool(poolA, hostA, nil)
-	if err != nil {
-		t.Fatalf("nodeA NewSyncedPool error: %v", err)
-	}
-	syncedB, err := NewSyncedPool(poolB, hostB, nil)
-	if err != nil {
-		t.Fatalf("nodeB NewSyncedPool error: %v", err)
-	}
-
-	// Start gRPC servers (CartActor + ControlPlane) on different ports.
-	grpcSrvA, err := StartGRPCServer(addrA, poolA, syncedA)
-	if err != nil {
-		t.Fatalf("StartGRPCServer A error: %v", err)
-	}
-	defer grpcSrvA.GracefulStop()
-
-	grpcSrvB, err := StartGRPCServer(addrB, poolB, syncedB)
-	if err != nil {
-		t.Fatalf("StartGRPCServer B error: %v", err)
-	}
-	defer grpcSrvB.GracefulStop()
-
-	// Helper to connect one pool to the other's server (manual AddRemote equivalent).
-	link := func(src *SyncedPool, remoteHost, remoteAddr string) {
-		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-		defer cancel()
-		conn, dialErr := grpc.DialContext(ctx, remoteAddr, grpc.WithInsecure(), grpc.WithBlock())
-		if dialErr != nil {
-			t.Fatalf("dial %s (%s) failed: %v", remoteHost, remoteAddr, dialErr)
-		}
-		cartClient := messages.NewCartActorClient(conn)
-		controlClient := messages.NewControlPlaneClient(conn)
-
-		src.mu.Lock()
-		src.remoteHosts[remoteHost] = &RemoteHostGRPC{
-			Host:          remoteHost,
-			Conn:          conn,
-			CartClient:    cartClient,
-			ControlClient: controlClient,
-		}
-		src.mu.Unlock()
-	}
-
-	// Cross-link the two pools.
-	link(syncedA, hostB, addrB)
-	link(syncedB, hostA, addrA)
-
-	// Rebuild rings after manual cross-link so deterministic ownership works immediately.
-	syncedA.ForceRingRefresh()
-	syncedB.ForceRingRefresh()
-
-	// Allow brief stabilization (control plane pings / no real negotiation needed here).
-	time.Sleep(200 * time.Millisecond)
-
-	// Create a deterministic cart id for test readability.
-	cartID := ToCartId(fmt.Sprintf("cart-%d", time.Now().UnixNano()))
-
-	// Mutation payload (ring-determined ownership; no assumption about which node owns).
-	addItem := &messages.AddItem{
-		ItemId:   1,
-		Quantity: 1,
-		Price:    1500,
-		OrgPrice: 1500,
-		Sku:      "sku-test-multi",
-		Name:     "Multi Node Test",
-		Image:    "/test.png",
-		Stock:    2,
-		Tax:      2500,
-		Country:  "se",
-	}
-
-	// Determine ring owner and set primary / secondary references.
-	ownerHost := syncedA.DebugOwnerHost(cartID)
-	var ownerSynced, otherSynced *SyncedPool
-	var ownerPool, otherPool *GrainLocalPool
-	switch ownerHost {
-	case hostA:
-		ownerSynced, ownerPool = syncedA, poolA
-		otherSynced, otherPool = syncedB, poolB
-	case hostB:
-		ownerSynced, ownerPool = syncedB, poolB
-		otherSynced, otherPool = syncedA, poolA
-	default:
-		t.Fatalf("unexpected ring owner %s (expected %s or %s)", ownerHost, hostA, hostB)
-	}
-
-	// Apply mutation on the ring-designated owner.
-	if _, err := ownerSynced.Apply(cartID, addItem); err != nil {
-		t.Fatalf("owner %s Apply addItem error: %v", ownerHost, err)
-	}
-
-	// Validate owner pool has the grain and the other does not.
-	if _, ok := ownerPool.GetGrains()[cartID]; !ok {
-		t.Fatalf("expected owner %s to have local grain", ownerHost)
-	}
-	if _, ok := otherPool.GetGrains()[cartID]; ok {
-		t.Fatalf("non-owner unexpectedly holds local grain")
-	}
-
-	// Prepare change mutation to be applied from the non-owner (should route remotely).
-	change := &messages.ChangeQuantity{
-		Id:       1, // line id after first AddItem
-		Quantity: 2,
-	}
-	// Apply remotely via the non-owner.
-	if _, err := otherSynced.Apply(cartID, change); err != nil {
-		t.Fatalf("non-owner remote Apply changeQuantity error: %v", err)
-	}
-
-	// Remote re-mutation already performed via otherSynced; removed duplicate block.
-
-	// NodeB local grain assertion:
-	// Only assert absence if nodeB is NOT the ring-designated owner. If nodeB is the owner,
-	// it is expected to have a local grain (previous generic ownership assertions already ran).
-	if ownerHost != hostB {
-		if _, local := poolB.GetGrains()[cartID]; local {
-			t.Fatalf("nodeB unexpectedly created local grain (ownership duplication)")
-		}
-	}
-
-	// Fetch state from nodeB to ensure we see updated quantity (2).
-	grainStateB, err := syncedB.Get(cartID)
-	if err != nil {
-		t.Fatalf("nodeB Get error: %v", err)
-	}
-	if len(grainStateB.Items) != 1 || grainStateB.Items[0].Quantity != 2 {
-		t.Fatalf("nodeB observed inconsistent state: items=%d qty=%d (expected 1 / 2)",
-			len(grainStateB.Items),
-			func() int {
-				if len(grainStateB.Items) == 0 {
-					return -1
-				}
-				return grainStateB.Items[0].Quantity
-			}(),
-		)
-	}
-
-	// Cross-check from nodeA (authoritative) to ensure state matches.
-	grainStateA, err := syncedA.Get(cartID)
-	if err != nil {
-		t.Fatalf("nodeA Get error: %v", err)
-	}
-	if grainStateA.Items[0].Quantity != 2 {
-		t.Fatalf("nodeA authoritative state mismatch: expected qty=2 got %d", grainStateA.Items[0].Quantity)
-	}
-}

multi_node_three_test.go

@@ -1,304 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"testing"
-	"time"
-
-	messages "git.tornberg.me/go-cart-actor/proto"
-	"google.golang.org/grpc"
-)
-
-// TestThreeNodeMajorityOwnership validates ring-determined ownership and routing
-// in a 3-node cluster (A,B,C) using the consistent hashing ring (no quorum RPC).
-// The previous ConfirmOwner / quorum semantics have been removed; ownership is
-// deterministic and derived from the ring.
-//
-// It validates:
-//  1. The ring selects exactly one primary owner for a new cart.
-//  2. Other nodes (B,C) do NOT create local grains for the cart.
-//  3. Remote proxies are installed lazily so remote mutations can route.
-//  4. A remote mutation from one non-owner updates state visible on another.
-//  5. Authoritative state on the owner matches remote observations.
-//  6. (Future) This scaffolds replication tests when RF>1 is enabled.
-//
-// (Legacy comments about ConfirmOwner acceptance thresholds have been removed.)
-// (Function name retained for historical continuity.)
-func TestThreeNodeMajorityOwnership(t *testing.T) {
-	const (
-		addrA = "127.0.0.1:18181"
-		addrB = "127.0.0.1:18182"
-		addrC = "127.0.0.1:18183"
-		hostA = "nodeA3"
-		hostB = "nodeB3"
-		hostC = "nodeC3"
-	)
-
-	// Local grain pools
-	poolA := NewGrainLocalPool(1024, time.Minute, spawn)
-	poolB := NewGrainLocalPool(1024, time.Minute, spawn)
-	poolC := NewGrainLocalPool(1024, time.Minute, spawn)
-
-	// Synced pools (no discovery)
-	syncedA, err := NewSyncedPool(poolA, hostA, nil)
-	if err != nil {
-		t.Fatalf("nodeA NewSyncedPool error: %v", err)
-	}
-	syncedB, err := NewSyncedPool(poolB, hostB, nil)
-	if err != nil {
-		t.Fatalf("nodeB NewSyncedPool error: %v", err)
-	}
-	syncedC, err := NewSyncedPool(poolC, hostC, nil)
-	if err != nil {
-		t.Fatalf("nodeC NewSyncedPool error: %v", err)
-	}
-
-	// Start gRPC servers
-	grpcSrvA, err := StartGRPCServer(addrA, poolA, syncedA)
-	if err != nil {
-		t.Fatalf("StartGRPCServer A error: %v", err)
-	}
-	defer grpcSrvA.GracefulStop()
-	grpcSrvB, err := StartGRPCServer(addrB, poolB, syncedB)
-	if err != nil {
-		t.Fatalf("StartGRPCServer B error: %v", err)
-	}
-	defer grpcSrvB.GracefulStop()
-	grpcSrvC, err := StartGRPCServer(addrC, poolC, syncedC)
-	if err != nil {
-		t.Fatalf("StartGRPCServer C error: %v", err)
-	}
-	defer grpcSrvC.GracefulStop()
-
-	// Helper for manual cross-link (since AddRemote assumes fixed port)
-	link := func(src *SyncedPool, remoteHost, remoteAddr string) {
-		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-		defer cancel()
-		conn, dialErr := grpc.DialContext(ctx, remoteAddr, grpc.WithInsecure(), grpc.WithBlock())
-		if dialErr != nil {
-			t.Fatalf("dial %s (%s) failed: %v", remoteHost, remoteAddr, dialErr)
-		}
-		cartClient := messages.NewCartActorClient(conn)
-		controlClient := messages.NewControlPlaneClient(conn)
-
-		src.mu.Lock()
-		src.remoteHosts[remoteHost] = &RemoteHostGRPC{
-			Host:          remoteHost,
-			Conn:          conn,
-			CartClient:    cartClient,
-			ControlClient: controlClient,
-		}
-		src.mu.Unlock()
-	}
-
-	// Full mesh (each node knows all others)
-	link(syncedA, hostB, addrB)
-	link(syncedA, hostC, addrC)
-
-	link(syncedB, hostA, addrA)
-	link(syncedB, hostC, addrC)
-
-	link(syncedC, hostA, addrA)
-	link(syncedC, hostB, addrB)
-
-	// Rebuild rings after manual linking so ownership resolution is immediate.
-	syncedA.ForceRingRefresh()
-	syncedB.ForceRingRefresh()
-	syncedC.ForceRingRefresh()
-
-	// Allow brief stabilization
-	time.Sleep(200 * time.Millisecond)
-
-	// Deterministic-ish cart id
-	cartID := ToCartId(fmt.Sprintf("cart3-%d", time.Now().UnixNano()))
-
-	addItem := &messages.AddItem{
-		ItemId:   10,
-		Quantity: 1,
-		Price:    5000,
-		OrgPrice: 5000,
-		Sku:      "sku-3node",
-		Name:     "Three Node Test",
-		Image:    "/t.png",
-		Stock:    10,
-		Tax:      2500,
-		Country:  "se",
-	}
-
-	// Determine ring-designated owner (may be any of the three hosts)
-	ownerPre := syncedA.DebugOwnerHost(cartID)
-	if ownerPre != hostA && ownerPre != hostB && ownerPre != hostC {
-		t.Fatalf("ring returned unexpected owner %s (not in set {%s,%s,%s})", ownerPre, hostA, hostB, hostC)
-	}
-	var ownerSynced *SyncedPool
-	var ownerPool *GrainLocalPool
-	switch ownerPre {
-	case hostA:
-		ownerSynced, ownerPool = syncedA, poolA
-	case hostB:
-		ownerSynced, ownerPool = syncedB, poolB
-	case hostC:
-		ownerSynced, ownerPool = syncedC, poolC
-	}
-	// Pick two distinct non-owner nodes for remote mutation assertions
-	var remote1Synced, remote2Synced *SyncedPool
-	switch ownerPre {
-	case hostA:
-		remote1Synced, remote2Synced = syncedB, syncedC
-	case hostB:
-		remote1Synced, remote2Synced = syncedA, syncedC
-	case hostC:
-		remote1Synced, remote2Synced = syncedA, syncedB
-	}
-
-	// Apply on the ring-designated owner
-	if _, err := ownerSynced.Apply(cartID, addItem); err != nil {
-		t.Fatalf("owner %s Apply addItem error: %v", ownerPre, err)
-	}
-
-	// Small wait for remote proxy spawn (ring ownership already deterministic)
-	time.Sleep(150 * time.Millisecond)
-
-	// Assert only nodeA has local grain
-	localCount := 0
-	if _, ok := poolA.GetGrains()[cartID]; ok {
-		localCount++
-	}
-	if _, ok := poolB.GetGrains()[cartID]; ok {
-		localCount++
-	}
-	if _, ok := poolC.GetGrains()[cartID]; ok {
-		localCount++
-	}
-	if localCount != 1 {
-		t.Fatalf("expected exactly 1 local grain, got %d", localCount)
-	}
-	if _, ok := ownerPool.GetGrains()[cartID]; !ok {
-		t.Fatalf("expected owner %s to hold local grain", ownerPre)
-	}
-
-	// Remote proxies may not pre-exist; first remote mutation will trigger SpawnRemoteGrain lazily.
-
-	// Issue remote mutation from one non-owner -> ChangeQuantity (increase)
-	change := &messages.ChangeQuantity{
-		Id:       1,
-		Quantity: 3,
-	}
-	if _, err := remote1Synced.Apply(cartID, change); err != nil {
-		t.Fatalf("remote mutation (remote1) changeQuantity error: %v", err)
-	}
-
-	// Validate updated state visible via nodeC
-	stateC, err := remote2Synced.Get(cartID)
-	if err != nil {
-		t.Fatalf("nodeC Get error: %v", err)
-	}
-	if len(stateC.Items) != 1 || stateC.Items[0].Quantity != 3 {
-		t.Fatalf("nodeC observed state mismatch: items=%d qty=%d (expected 1 / 3)",
-			len(stateC.Items),
-			func() int {
-				if len(stateC.Items) == 0 {
-					return -1
-				}
-				return stateC.Items[0].Quantity
-			}(),
-		)
-	}
-
-	// Cross-check authoritative nodeA
-	stateA, err := syncedA.Get(cartID)
-	if err != nil {
-		t.Fatalf("nodeA Get error: %v", err)
-	}
-	if stateA.Items[0].Quantity != 3 {
-		t.Fatalf("nodeA authoritative state mismatch: expected qty=3 got %d", stateA.Items[0].Quantity)
-	}
-}
-
-// TestThreeNodeDiscoveryMajorityOwnership (placeholder)
-// This test is a scaffold demonstrating how a MockDiscovery would be wired
-// once AddRemote supports host:port (currently hard-coded to :1337).
-// It is skipped to avoid flakiness / false negatives until the production
-// AddRemote logic is enhanced to parse dynamic ports or the test harness
-// provides consistent port mapping.
-func TestThreeNodeDiscoveryMajorityOwnership(t *testing.T) {
-	t.Skip("Pending enhancement: AddRemote needs host:port support to fully exercise discovery-based multi-node linking")
-	// Example skeleton (non-functional with current AddRemote implementation):
-	//
-	// md := NewMockDiscovery([]string{"nodeB3", "nodeC3"})
-	// poolA := NewGrainLocalPool(1024, time.Minute, spawn)
-	// syncedA, err := NewSyncedPool(poolA, "nodeA3", md)
-	// if err != nil {
-	//     t.Fatalf("NewSyncedPool with mock discovery error: %v", err)
-	// }
-	// // Start server for nodeA (would also need servers for nodeB3/nodeC3 on expected ports)
-	// // grpcSrvA, _ := StartGRPCServer(":1337", poolA, syncedA)
-	// // defer grpcSrvA.GracefulStop()
-	//
-	// // Dynamically add a host via discovery
-	// // md.AddHost("nodeB3")
-	// // time.Sleep(100 * time.Millisecond) // allow AddRemote attempt
-	//
-	// // Assertions would verify syncedA.remoteHosts contains "nodeB3"
-}
-
-// TestHostRemovalAndErrorWithMockDiscovery validates behavior when:
-// 1. Discovery reports a host that cannot be dialed (AddRemote error path)
-// 2. That host is then removed (Deleted event) without leaving residual state
-// 3. A second failing host is added afterward (ensuring watcher still processes events)
-//
-// NOTE: Because AddRemote currently hard-codes :1337 and we are NOT starting a
-// real server for the bogus hosts, the dial will fail and the remote host should
-// never appear in remoteHosts. This intentionally exercises the error logging
-// path: "AddRemote: dial ... failed".
-func TestHostRemovalAndErrorWithMockDiscovery(t *testing.T) {
-	// Start a real node A (acts as the observing node)
-	const addrA = "127.0.0.1:18281"
-	hostA := "nodeA-md"
-
-	poolA := NewGrainLocalPool(128, time.Minute, spawn)
-
-	// Mock discovery starts with one bogus host that will fail to connect.
-	md := NewMockDiscovery([]string{"bogus-host-1"})
-	syncedA, err := NewSyncedPool(poolA, hostA, md)
-	if err != nil {
-		t.Fatalf("NewSyncedPool error: %v", err)
-	}
-
-	grpcSrvA, err := StartGRPCServer(addrA, poolA, syncedA)
-	if err != nil {
-		t.Fatalf("StartGRPCServer A error: %v", err)
-	}
-	defer grpcSrvA.GracefulStop()
-
-	// Kick off watch processing by starting Watch() (NewSyncedPool does this internally
-	// when discovery is non-nil, but we ensure events channel is active).
-	// The initial bogus host should trigger AddRemote -> dial failure.
-	time.Sleep(300 * time.Millisecond)
-
-	syncedA.mu.RLock()
-	if len(syncedA.remoteHosts) != 0 {
-		syncedA.mu.RUnlock()
-		t.Fatalf("expected 0 remoteHosts after failing dial, got %d", len(syncedA.remoteHosts))
-	}
-	syncedA.mu.RUnlock()
-
-	// Remove the bogus host (should not panic; no entry to clean up).
-	md.RemoveHost("bogus-host-1")
-	time.Sleep(100 * time.Millisecond)
-
-	// Add another bogus host to ensure watcher still alive.
-	md.AddHost("bogus-host-2")
-	time.Sleep(300 * time.Millisecond)
-
-	syncedA.mu.RLock()
-	if len(syncedA.remoteHosts) != 0 {
-		syncedA.mu.RUnlock()
-		t.Fatalf("expected 0 remoteHosts after second failing dial, got %d", len(syncedA.remoteHosts))
-	}
-	syncedA.mu.RUnlock()
-
-	// Clean up discovery
-	md.Close()
-}

mutation_add_item.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_add_item.go

mutation_add_request.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_add_request.go

mutation_change_quantity.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_change_quantity.go

mutation_initialize_checkout.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_initialize_checkout.go

mutation_order_created.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_order_created.go

mutation_remove_delivery.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_remove_delivery.go

mutation_remove_item.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_remove_item.go

mutation_set_cart_items.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_set_cart_items.go

mutation_set_delivery.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"slices"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_set_delivery.go

mutation_set_pickup_point.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 // mutation_set_pickup_point.go

pkg/actor/grain.go

@@ -0,0 +1,11 @@
+package actor
+
+import "time"
+
+type Grain[V any] interface {
+	GetId() uint64
+	Apply(content any, isReplay bool) (*V, error)
+	GetLastAccess() time.Time
+	GetLastChange() time.Time
+	GetCurrentState() (*V, error)
+}

pkg/actor/grain_pool.go

@@ -0,0 +1,25 @@
+package actor
+
+import "net/http"
+
+type GrainPool[V any] interface {
+	Apply(id uint64, mutation any) (V, error)
+	Get(id uint64) (V, error)
+	OwnerHost(id uint64) (Host, bool)
+	Hostname() string
+	TakeOwnership(id uint64)
+	HandleOwnershipChange(host string, ids []uint64) error
+	HandleRemoteExpiry(host string, ids []uint64) error
+	Negotiate(otherHosts []string)
+	GetLocalIds() []uint64
+	RemoveHost(host string)
+	IsHealthy() bool
+	Close()
+}
+
+// Host abstracts a remote node capable of proxying cart requests.
+type Host interface {
+	Name() string
+	Proxy(id uint64, w http.ResponseWriter, r *http.Request) (bool, error)
+	GetActorIds() []uint64
+}

pkg/actor/grpc_server.go

@@ -0,0 +1,102 @@
+package actor
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net"
+	"time"
+
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/reflection"
+)
+
+// ControlServer implements the ControlPlane gRPC services.
+// It delegates to a grain pool and cluster operations to a synced pool.
+type ControlServer[V any] struct {
+	messages.UnimplementedControlPlaneServer
+
+	pool GrainPool[V]
+}
+
+func (s *ControlServer[V]) AnnounceOwnership(ctx context.Context, req *messages.OwnershipAnnounce) (*messages.OwnerChangeAck, error) {
+	err := s.pool.HandleOwnershipChange(req.Host, req.Ids)
+	if err != nil {
+		return &messages.OwnerChangeAck{
+			Accepted: false,
+			Message:  "owner change failed",
+		}, err
+	}
+
+	log.Printf("Ack count: %d", len(req.Ids))
+	return &messages.OwnerChangeAck{
+		Accepted: true,
+		Message:  "ownership announced",
+	}, nil
+}
+
+func (s *ControlServer[V]) AnnounceExpiry(ctx context.Context, req *messages.ExpiryAnnounce) (*messages.OwnerChangeAck, error) {
+	err := s.pool.HandleRemoteExpiry(req.Host, req.Ids)
+	return &messages.OwnerChangeAck{
+		Accepted: err == nil,
+		Message:  "expiry acknowledged",
+	}, nil
+}
+
+// ControlPlane: Ping
+func (s *ControlServer[V]) Ping(ctx context.Context, _ *messages.Empty) (*messages.PingReply, error) {
+
+	return &messages.PingReply{
+		Host:     s.pool.Hostname(),
+		UnixTime: time.Now().Unix(),
+	}, nil
+}
+
+// ControlPlane: Negotiate (merge host views)
+func (s *ControlServer[V]) Negotiate(ctx context.Context, req *messages.NegotiateRequest) (*messages.NegotiateReply, error) {
+
+	s.pool.Negotiate(req.KnownHosts)
+	return &messages.NegotiateReply{Hosts: req.GetKnownHosts()}, nil
+}
+
+// ControlPlane: GetCartIds (locally owned carts only)
+func (s *ControlServer[V]) GetLocalActorIds(ctx context.Context, _ *messages.Empty) (*messages.ActorIdsReply, error) {
+	return &messages.ActorIdsReply{Ids: s.pool.GetLocalIds()}, nil
+}
+
+// ControlPlane: Closing (peer shutdown notification)
+func (s *ControlServer[V]) Closing(ctx context.Context, req *messages.ClosingNotice) (*messages.OwnerChangeAck, error) {
+	if req.GetHost() != "" {
+		s.pool.RemoveHost(req.GetHost())
+	}
+	return &messages.OwnerChangeAck{
+		Accepted: true,
+		Message:  "removed host",
+	}, nil
+}
+
+// StartGRPCServer configures and starts the unified gRPC server on the given address.
+// It registers both the CartActor and ControlPlane services.
+func NewControlServer[V any](addr string, pool GrainPool[V]) (*grpc.Server, error) {
+	lis, err := net.Listen("tcp", addr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to listen: %w", err)
+	}
+
+	grpcServer := grpc.NewServer()
+	server := &ControlServer[V]{
+		pool: pool,
+	}
+
+	log.Printf("gRPC server listening on %s", addr)
+
+	if err := grpcServer.Serve(lis); err != nil {
+		log.Fatalf("failed to serve gRPC: %v", err)
+	}
+
+	messages.RegisterControlPlaneServer(grpcServer, server)
+	reflection.Register(grpcServer)
+
+	return grpcServer, nil
+}

pkg/discovery/discovery.go

@@ -0,0 +1,72 @@
+package discovery
+
+import (
+	"context"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+	toolsWatch "k8s.io/client-go/tools/watch"
+)
+
+type K8sDiscovery struct {
+	ctx    context.Context
+	client *kubernetes.Clientset
+}
+
+func (k *K8sDiscovery) Discover() ([]string, error) {
+	return k.DiscoverInNamespace("")
+}
+func (k *K8sDiscovery) DiscoverInNamespace(namespace string) ([]string, error) {
+	pods, err := k.client.CoreV1().Pods(namespace).List(k.ctx, metav1.ListOptions{
+		LabelSelector: "actor-pool=cart",
+	})
+	if err != nil {
+		return nil, err
+	}
+	hosts := make([]string, 0, len(pods.Items))
+	for _, pod := range pods.Items {
+		hosts = append(hosts, pod.Status.PodIP)
+	}
+	return hosts, nil
+}
+
+type HostChange struct {
+	Host string
+	Type watch.EventType
+}
+
+func (k *K8sDiscovery) Watch() (<-chan HostChange, error) {
+	timeout := int64(30)
+	watcherFn := func(options metav1.ListOptions) (watch.Interface, error) {
+		return k.client.CoreV1().Pods("").Watch(k.ctx, metav1.ListOptions{
+			LabelSelector:  "actor-pool=cart",
+			TimeoutSeconds: &timeout,
+		})
+	}
+	watcher, err := toolsWatch.NewRetryWatcher("1", &cache.ListWatch{WatchFunc: watcherFn})
+	if err != nil {
+		return nil, err
+	}
+	ch := make(chan HostChange)
+	go func() {
+		for event := range watcher.ResultChan() {
+
+			pod := event.Object.(*v1.Pod)
+			ch <- HostChange{
+				Host: pod.Status.PodIP,
+				Type: event.Type,
+			}
+		}
+	}()
+	return ch, nil
+}
+
+func NewK8sDiscovery(client *kubernetes.Clientset) *K8sDiscovery {
+	return &K8sDiscovery{
+		ctx:    context.Background(),
+		client: client,
+	}
+}

pkg/discovery/discovery_mock.go

@@ -1,82 +1,12 @@
-package main
+package discovery
 
 import (
 	"context"
 	"sync"
 
-	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/watch"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/tools/cache"
-	toolsWatch "k8s.io/client-go/tools/watch"
 )
 
-type Discovery interface {
-	Discover() ([]string, error)
-	Watch() (<-chan HostChange, error)
-}
-
-type K8sDiscovery struct {
-	ctx    context.Context
-	client *kubernetes.Clientset
-}
-
-func (k *K8sDiscovery) Discover() ([]string, error) {
-	return k.DiscoverInNamespace("")
-}
-func (k *K8sDiscovery) DiscoverInNamespace(namespace string) ([]string, error) {
-	pods, err := k.client.CoreV1().Pods(namespace).List(k.ctx, metav1.ListOptions{
-		LabelSelector: "actor-pool=cart",
-	})
-	if err != nil {
-		return nil, err
-	}
-	hosts := make([]string, 0, len(pods.Items))
-	for _, pod := range pods.Items {
-		hosts = append(hosts, pod.Status.PodIP)
-	}
-	return hosts, nil
-}
-
-type HostChange struct {
-	Host string
-	Type watch.EventType
-}
-
-func (k *K8sDiscovery) Watch() (<-chan HostChange, error) {
-	timeout := int64(30)
-	watcherFn := func(options metav1.ListOptions) (watch.Interface, error) {
-		return k.client.CoreV1().Pods("").Watch(k.ctx, metav1.ListOptions{
-			LabelSelector:  "actor-pool=cart",
-			TimeoutSeconds: &timeout,
-		})
-	}
-	watcher, err := toolsWatch.NewRetryWatcher("1", &cache.ListWatch{WatchFunc: watcherFn})
-	if err != nil {
-		return nil, err
-	}
-	ch := make(chan HostChange)
-	go func() {
-		for event := range watcher.ResultChan() {
-
-			pod := event.Object.(*v1.Pod)
-			ch <- HostChange{
-				Host: pod.Status.PodIP,
-				Type: event.Type,
-			}
-		}
-	}()
-	return ch, nil
-}
-
-func NewK8sDiscovery(client *kubernetes.Clientset) *K8sDiscovery {
-	return &K8sDiscovery{
-		ctx:    context.Background(),
-		client: client,
-	}
-}
-
 // MockDiscovery is an in-memory Discovery implementation for tests.
 // It allows deterministic injection of host additions/removals without
 // depending on Kubernetes API machinery.

pkg/discovery/discovery_test.go

@@ -1,4 +1,4 @@
-package main
+package discovery
 
 import (
 	"testing"

pkg/discovery/types.go

@@ -0,0 +1,6 @@
+package discovery
+
+type Discovery interface {
+	Discover() ([]string, error)
+	Watch() (<-chan HostChange, error)
+}

pkg/messages/control_plane.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.36.10
-// 	protoc        v3.21.12
+// 	protoc        v6.32.1
 // source: control_plane.proto
 
 package messages
@@ -202,27 +202,27 @@ func (x *NegotiateReply) GetHosts() []string {
 }
 
 // CartIdsReply returns the list of cart IDs (string form) currently owned locally.
-type CartIdsReply struct {
+type ActorIdsReply struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
-	CartIds       []string               `protobuf:"bytes,1,rep,name=cart_ids,json=cartIds,proto3" json:"cart_ids,omitempty"`
+	Ids           []uint64               `protobuf:"varint,1,rep,packed,name=ids,proto3" json:"ids,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
 
-func (x *CartIdsReply) Reset() {
-	*x = CartIdsReply{}
+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 *CartIdsReply) String() string {
+func (x *ActorIdsReply) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*CartIdsReply) ProtoMessage() {}
+func (*ActorIdsReply) ProtoMessage() {}
 
-func (x *CartIdsReply) ProtoReflect() protoreflect.Message {
+func (x *ActorIdsReply) ProtoReflect() protoreflect.Message {
 	mi := &file_control_plane_proto_msgTypes[4]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -234,14 +234,14 @@ func (x *CartIdsReply) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use CartIdsReply.ProtoReflect.Descriptor instead.
-func (*CartIdsReply) Descriptor() ([]byte, []int) {
+// Deprecated: Use ActorIdsReply.ProtoReflect.Descriptor instead.
+func (*ActorIdsReply) Descriptor() ([]byte, []int) {
 	return file_control_plane_proto_rawDescGZIP(), []int{4}
 }
 
-func (x *CartIdsReply) GetCartIds() []string {
+func (x *ActorIdsReply) GetIds() []uint64 {
 	if x != nil {
-		return x.CartIds
+		return x.Ids
 	}
 	return nil
 }
@@ -344,6 +344,113 @@ func (x *ClosingNotice) GetHost() string {
 	return ""
 }
 
+// OwnershipAnnounce broadcasts first-touch ownership claims for cart IDs.
+// First claim wins; receivers SHOULD NOT overwrite an existing different owner.
+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
+	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
+}
+
+// ExpiryAnnounce broadcasts that a host evicted the provided cart IDs.
+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"`
+	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
+}
+
 var File_control_plane_proto protoreflect.FileDescriptor
 
 const file_control_plane_proto_rawDesc = "" +
@@ -357,19 +464,26 @@ const file_control_plane_proto_rawDesc = "" +
 	"\vknown_hosts\x18\x01 \x03(\tR\n" +
 	"knownHosts\"&\n" +
 	"\x0eNegotiateReply\x12\x14\n" +
-	"\x05hosts\x18\x01 \x03(\tR\x05hosts\")\n" +
-	"\fCartIdsReply\x12\x19\n" +
-	"\bcart_ids\x18\x01 \x03(\tR\acartIds\"F\n" +
+	"\x05hosts\x18\x01 \x03(\tR\x05hosts\"!\n" +
+	"\rActorIdsReply\x12\x10\n" +
+	"\x03ids\x18\x01 \x03(\x04R\x03ids\"F\n" +
 	"\x0eOwnerChangeAck\x12\x1a\n" +
 	"\baccepted\x18\x01 \x01(\bR\baccepted\x12\x18\n" +
 	"\amessage\x18\x02 \x01(\tR\amessage\"#\n" +
 	"\rClosingNotice\x12\x12\n" +
-	"\x04host\x18\x01 \x01(\tR\x04host2\xf4\x01\n" +
+	"\x04host\x18\x01 \x01(\tR\x04host\"9\n" +
+	"\x11OwnershipAnnounce\x12\x12\n" +
+	"\x04host\x18\x01 \x01(\tR\x04host\x12\x10\n" +
+	"\x03ids\x18\x02 \x03(\x04R\x03ids\"6\n" +
+	"\x0eExpiryAnnounce\x12\x12\n" +
+	"\x04host\x18\x01 \x01(\tR\x04host\x12\x10\n" +
+	"\x03ids\x18\x02 \x03(\x04R\x03ids2\x8d\x03\n" +
 	"\fControlPlane\x12,\n" +
 	"\x04Ping\x12\x0f.messages.Empty\x1a\x13.messages.PingReply\x12A\n" +
-	"\tNegotiate\x12\x1a.messages.NegotiateRequest\x1a\x18.messages.NegotiateReply\x125\n" +
-	"\n" +
-	"GetCartIds\x12\x0f.messages.Empty\x1a\x16.messages.CartIdsReply\x12<\n" +
+	"\tNegotiate\x12\x1a.messages.NegotiateRequest\x1a\x18.messages.NegotiateReply\x12<\n" +
+	"\x10GetLocalActorIds\x12\x0f.messages.Empty\x1a\x17.messages.ActorIdsReply\x12J\n" +
+	"\x11AnnounceOwnership\x12\x1b.messages.OwnershipAnnounce\x1a\x18.messages.OwnerChangeAck\x12D\n" +
+	"\x0eAnnounceExpiry\x12\x18.messages.ExpiryAnnounce\x1a\x18.messages.OwnerChangeAck\x12<\n" +
 	"\aClosing\x12\x17.messages.ClosingNotice\x1a\x18.messages.OwnerChangeAckB.Z,git.tornberg.me/go-cart-actor/proto;messagesb\x06proto3"
 
 var (
@@ -384,27 +498,33 @@ func file_control_plane_proto_rawDescGZIP() []byte {
 	return file_control_plane_proto_rawDescData
 }
 
-var file_control_plane_proto_msgTypes = make([]protoimpl.MessageInfo, 7)
+var file_control_plane_proto_msgTypes = make([]protoimpl.MessageInfo, 9)
 var file_control_plane_proto_goTypes = []any{
 	(*Empty)(nil),             // 0: messages.Empty
 	(*PingReply)(nil),         // 1: messages.PingReply
 	(*NegotiateRequest)(nil),  // 2: messages.NegotiateRequest
 	(*NegotiateReply)(nil),    // 3: messages.NegotiateReply
-	(*CartIdsReply)(nil),     // 4: messages.CartIdsReply
+	(*ActorIdsReply)(nil),     // 4: messages.ActorIdsReply
 	(*OwnerChangeAck)(nil),    // 5: messages.OwnerChangeAck
 	(*ClosingNotice)(nil),     // 6: messages.ClosingNotice
+	(*OwnershipAnnounce)(nil), // 7: messages.OwnershipAnnounce
+	(*ExpiryAnnounce)(nil),    // 8: messages.ExpiryAnnounce
 }
 var file_control_plane_proto_depIdxs = []int32{
 	0, // 0: messages.ControlPlane.Ping:input_type -> messages.Empty
 	2, // 1: messages.ControlPlane.Negotiate:input_type -> messages.NegotiateRequest
-	0, // 2: messages.ControlPlane.GetCartIds:input_type -> messages.Empty
-	6, // 3: messages.ControlPlane.Closing:input_type -> messages.ClosingNotice
-	1, // 4: messages.ControlPlane.Ping:output_type -> messages.PingReply
-	3, // 5: messages.ControlPlane.Negotiate:output_type -> messages.NegotiateReply
-	4, // 6: messages.ControlPlane.GetCartIds:output_type -> messages.CartIdsReply
-	5, // 7: messages.ControlPlane.Closing:output_type -> messages.OwnerChangeAck
-	4, // [4:8] is the sub-list for method output_type
-	0, // [0:4] is the sub-list for method input_type
+	0, // 2: messages.ControlPlane.GetLocalActorIds:input_type -> messages.Empty
+	7, // 3: messages.ControlPlane.AnnounceOwnership:input_type -> messages.OwnershipAnnounce
+	8, // 4: messages.ControlPlane.AnnounceExpiry:input_type -> messages.ExpiryAnnounce
+	6, // 5: messages.ControlPlane.Closing:input_type -> messages.ClosingNotice
+	1, // 6: messages.ControlPlane.Ping:output_type -> messages.PingReply
+	3, // 7: messages.ControlPlane.Negotiate:output_type -> messages.NegotiateReply
+	4, // 8: messages.ControlPlane.GetLocalActorIds:output_type -> messages.ActorIdsReply
+	5, // 9: messages.ControlPlane.AnnounceOwnership:output_type -> messages.OwnerChangeAck
+	5, // 10: messages.ControlPlane.AnnounceExpiry:output_type -> messages.OwnerChangeAck
+	5, // 11: messages.ControlPlane.Closing:output_type -> messages.OwnerChangeAck
+	6, // [6:12] is the sub-list for method output_type
+	0, // [0:6] is the sub-list for method input_type
 	0, // [0:0] is the sub-list for extension type_name
 	0, // [0:0] is the sub-list for extension extendee
 	0, // [0:0] is the sub-list for field type_name
@@ -421,7 +541,7 @@ func file_control_plane_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_control_plane_proto_rawDesc), len(file_control_plane_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   7,
+			NumMessages:   9,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

pkg/messages/control_plane_grpc.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.5.1
-// - protoc             v3.21.12
+// - protoc             v6.32.1
 // source: control_plane.proto
 
 package messages
@@ -21,7 +21,9 @@ const _ = grpc.SupportPackageIsVersion9
 const (
 	ControlPlane_Ping_FullMethodName              = "/messages.ControlPlane/Ping"
 	ControlPlane_Negotiate_FullMethodName         = "/messages.ControlPlane/Negotiate"
-	ControlPlane_GetCartIds_FullMethodName = "/messages.ControlPlane/GetCartIds"
+	ControlPlane_GetLocalActorIds_FullMethodName  = "/messages.ControlPlane/GetLocalActorIds"
+	ControlPlane_AnnounceOwnership_FullMethodName = "/messages.ControlPlane/AnnounceOwnership"
+	ControlPlane_AnnounceExpiry_FullMethodName    = "/messages.ControlPlane/AnnounceExpiry"
 	ControlPlane_Closing_FullMethodName           = "/messages.ControlPlane/Closing"
 )
 
@@ -36,7 +38,11 @@ type ControlPlaneClient interface {
 	// 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.
-	GetCartIds(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*CartIdsReply, error)
+	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)
+	// 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)
 }
@@ -69,10 +75,30 @@ func (c *controlPlaneClient) Negotiate(ctx context.Context, in *NegotiateRequest
 	return out, nil
 }
 
-func (c *controlPlaneClient) GetCartIds(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*CartIdsReply, error) {
+func (c *controlPlaneClient) GetLocalActorIds(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*ActorIdsReply, error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartIdsReply)
-	err := c.cc.Invoke(ctx, ControlPlane_GetCartIds_FullMethodName, in, out, cOpts...)
+	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) 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
 	}
@@ -100,7 +126,11 @@ type ControlPlaneServer interface {
 	// Negotiate merges host views; used during discovery & convergence.
 	Negotiate(context.Context, *NegotiateRequest) (*NegotiateReply, error)
 	// GetCartIds lists currently owned cart IDs on this node.
-	GetCartIds(context.Context, *Empty) (*CartIdsReply, error)
+	GetLocalActorIds(context.Context, *Empty) (*ActorIdsReply, error)
+	// Ownership announcement: first-touch claim broadcast (idempotent; best-effort).
+	AnnounceOwnership(context.Context, *OwnershipAnnounce) (*OwnerChangeAck, 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)
 	mustEmbedUnimplementedControlPlaneServer()
@@ -119,8 +149,14 @@ func (UnimplementedControlPlaneServer) Ping(context.Context, *Empty) (*PingReply
 func (UnimplementedControlPlaneServer) Negotiate(context.Context, *NegotiateRequest) (*NegotiateReply, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method Negotiate not implemented")
 }
-func (UnimplementedControlPlaneServer) GetCartIds(context.Context, *Empty) (*CartIdsReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method GetCartIds 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) 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")
@@ -182,20 +218,56 @@ func _ControlPlane_Negotiate_Handler(srv interface{}, ctx context.Context, dec f
 	return interceptor(ctx, in, info, handler)
 }
 
-func _ControlPlane_GetCartIds_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+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).GetCartIds(ctx, in)
+		return srv.(ControlPlaneServer).GetLocalActorIds(ctx, in)
 	}
 	info := &grpc.UnaryServerInfo{
 		Server:     srv,
-		FullMethod: ControlPlane_GetCartIds_FullMethodName,
+		FullMethod: ControlPlane_GetLocalActorIds_FullMethodName,
 	}
 	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(ControlPlaneServer).GetCartIds(ctx, req.(*Empty))
+		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_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)
 }
@@ -234,8 +306,16 @@ var ControlPlane_ServiceDesc = grpc.ServiceDesc{
 			Handler:    _ControlPlane_Negotiate_Handler,
 		},
 		{
-			MethodName: "GetCartIds",
-			Handler:    _ControlPlane_GetCartIds_Handler,
+			MethodName: "GetLocalActorIds",
+			Handler:    _ControlPlane_GetLocalActorIds_Handler,
+		},
+		{
+			MethodName: "AnnounceOwnership",
+			Handler:    _ControlPlane_AnnounceOwnership_Handler,
+		},
+		{
+			MethodName: "AnnounceExpiry",
+			Handler:    _ControlPlane_AnnounceExpiry_Handler,
 		},
 		{
 			MethodName: "Closing",

pkg/messages/messages.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.36.10
-// 	protoc        v3.21.12
+// 	protoc        v6.32.1
 // source: messages.proto
 
 package messages

pkg/proxy/remotehost.go

@@ -0,0 +1,209 @@
+package proxy
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"time"
+
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// RemoteHost mirrors the lightweight controller used for remote node
+// interaction.
+type RemoteHost struct {
+	Host          string
+	httpBase      string
+	conn          *grpc.ClientConn
+	transport     *http.Transport
+	client        *http.Client
+	controlClient messages.ControlPlaneClient
+	MissedPings   int
+}
+
+func NewRemoteHost(host string) (*RemoteHost, error) {
+
+	target := fmt.Sprintf("%s:1337", host)
+
+	conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(insecure.NewCredentials()))
+
+	if err != nil {
+		log.Printf("AddRemote: dial %s failed: %v", target, err)
+		return nil, err
+	}
+
+	controlClient := messages.NewControlPlaneClient(conn)
+	for retries := range 3 {
+		ctx, pingCancel := context.WithTimeout(context.Background(), time.Second)
+		_, pingErr := controlClient.Ping(ctx, &messages.Empty{})
+		pingCancel()
+		if pingErr == nil {
+			break
+		}
+		if retries == 2 {
+			log.Printf("AddRemote: ping %s failed after retries: %v", host, pingErr)
+			conn.Close()
+			return nil, pingErr
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+
+	transport := &http.Transport{
+		MaxIdleConns:        100,
+		MaxIdleConnsPerHost: 100,
+		IdleConnTimeout:     120 * time.Second,
+	}
+	client := &http.Client{Transport: transport, Timeout: 10 * time.Second}
+
+	return &RemoteHost{
+		Host:          host,
+		httpBase:      fmt.Sprintf("http://%s:8080/cart", host),
+		conn:          conn,
+		transport:     transport,
+		client:        client,
+		controlClient: controlClient,
+		MissedPings:   0,
+	}, nil
+}
+
+func (h *RemoteHost) Name() string {
+	return h.Host
+}
+
+func (h *RemoteHost) Close() error {
+	if h.conn != nil {
+		h.conn.Close()
+	}
+	return nil
+}
+
+func (h *RemoteHost) Ping() bool {
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	_, err := h.controlClient.Ping(ctx, &messages.Empty{})
+	cancel()
+	if err != nil {
+		h.MissedPings++
+		log.Printf("Ping %s failed (%d) %v", h.Host, h.MissedPings, err)
+		return false
+	}
+
+	h.MissedPings = 0
+	return true
+}
+
+func (h *RemoteHost) Negotiate(knownHosts []string) ([]string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	resp, err := h.controlClient.Negotiate(ctx, &messages.NegotiateRequest{
+		KnownHosts: knownHosts,
+	})
+	if err != nil {
+		h.MissedPings++
+		log.Printf("Negotiate %s failed: %v", h.Host, err)
+		return nil, err
+	}
+	h.MissedPings = 0
+	return resp.Hosts, nil
+}
+
+func (h *RemoteHost) GetActorIds() []uint64 {
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	reply, err := h.controlClient.GetLocalActorIds(ctx, &messages.Empty{})
+	if err != nil {
+		log.Printf("Init remote %s: GetCartIds error: %v", h.Host, err)
+		h.MissedPings++
+		return []uint64{}
+	}
+	return reply.GetIds()
+}
+
+func (h *RemoteHost) AnnounceOwnership(uids []uint64) {
+	_, err := h.controlClient.AnnounceOwnership(context.Background(), &messages.OwnershipAnnounce{
+		Host: h.Host,
+		Ids:  uids,
+	})
+	if err != nil {
+		log.Printf("ownership announce to %s failed: %v", h.Host, err)
+		h.MissedPings++
+		return
+	}
+	h.MissedPings = 0
+}
+
+func (h *RemoteHost) AnnounceExpiry(uids []uint64) {
+	_, err := h.controlClient.AnnounceExpiry(context.Background(), &messages.ExpiryAnnounce{
+		Host: h.Host,
+		Ids:  uids,
+	})
+	if err != nil {
+		log.Printf("expiry announce to %s failed: %v", h.Host, err)
+		h.MissedPings++
+		return
+	}
+	h.MissedPings = 0
+}
+
+func (h *RemoteHost) Proxy(id uint64, w http.ResponseWriter, r *http.Request) (bool, error) {
+	target := fmt.Sprintf("%s%s", h.httpBase, r.URL.RequestURI())
+	// var bodyCopy []byte
+	// if r.Body != nil && r.Body != http.NoBody {
+	// 	var err error
+	// 	bodyCopy, err = io.ReadAll(r.Body)
+	// 	if err != nil {
+	// 		http.Error(w, "proxy read error", http.StatusBadGateway)
+	// 		return false, err
+	// 	}
+	// }
+	// if r.Body != nil {
+	// 	r.Body.Close()
+	// }
+	// var reqBody io.Reader
+	// if len(bodyCopy) > 0 {
+	// 	reqBody = bytes.NewReader(bodyCopy)
+	// }
+	req, err := http.NewRequestWithContext(r.Context(), r.Method, target, r.Body)
+	if err != nil {
+		http.Error(w, "proxy build error", http.StatusBadGateway)
+		return false, err
+	}
+	//r.Body = io.NopCloser(bytes.NewReader(bodyCopy))
+	req.Header.Set("X-Forwarded-Host", r.Host)
+
+	for k, v := range r.Header {
+		for _, vv := range v {
+			req.Header.Add(k, vv)
+		}
+	}
+	res, err := h.client.Do(req)
+	if err != nil {
+		http.Error(w, "proxy request error", http.StatusBadGateway)
+		return false, err
+	}
+	defer res.Body.Close()
+	for k, v := range res.Header {
+		for _, vv := range v {
+			w.Header().Add(k, vv)
+		}
+	}
+	w.Header().Set("X-Cart-Owner-Routed", "true")
+	if res.StatusCode >= 200 && res.StatusCode <= 299 {
+		w.WriteHeader(res.StatusCode)
+		_, copyErr := io.Copy(w, res.Body)
+		if copyErr != nil {
+			return true, copyErr
+		}
+		return true, nil
+	}
+	return false, fmt.Errorf("proxy response status %d", res.StatusCode)
+}
+
+func (r *RemoteHost) IsHealthy() bool {
+	return r.MissedPings < 3
+}

pool-server.go

@@ -5,64 +5,48 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
-	"math/rand"
 	"net/http"
 	"strconv"
 	"time"
 
-	messages "git.tornberg.me/go-cart-actor/proto"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 )
 
 type PoolServer struct {
 	pod_name string
-	pool     GrainPool
+	pool     *CartPool
 }
 
-func NewPoolServer(pool GrainPool, pod_name string) *PoolServer {
+func NewPoolServer(pool *CartPool, pod_name string) *PoolServer {
 	return &PoolServer{
 		pod_name: pod_name,
 		pool:     pool,
 	}
 }
 
-func (s *PoolServer) process(id CartId, mutation interface{}) (*messages.CartState, error) {
-	grain, err := s.pool.Apply(id, mutation)
-	if err != nil {
-		return nil, err
-	}
-	return ToCartState(grain), nil
+func (s *PoolServer) ApplyLocal(id CartId, mutation interface{}) (*CartGrain, error) {
+	return s.pool.Apply(uint64(id), mutation)
 }
 
 func (s *PoolServer) HandleGet(w http.ResponseWriter, r *http.Request, id CartId) error {
-	grain, err := s.pool.Get(id)
+	grain, err := s.pool.Get(uint64(id))
 	if err != nil {
 		return err
 	}
 
-	return s.WriteResult(w, ToCartState(grain))
+	return s.WriteResult(w, grain)
 }
 
 func (s *PoolServer) HandleAddSku(w http.ResponseWriter, r *http.Request, id CartId) error {
 	sku := r.PathValue("sku")
-	data, err := s.process(id, &messages.AddRequest{Sku: sku, Quantity: 1})
+	data, err := s.ApplyLocal(id, &messages.AddRequest{Sku: sku, Quantity: 1})
 	if err != nil {
 		return err
 	}
 	return s.WriteResult(w, data)
 }
 
-func ErrorHandler(fn func(w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
-	return func(w http.ResponseWriter, r *http.Request) {
-		err := fn(w, r)
-		if err != nil {
-			log.Printf("Server error, not remote error: %v\n", err)
-			w.WriteHeader(http.StatusInternalServerError)
-			w.Write([]byte(err.Error()))
-		}
-	}
-}
-
-func (s *PoolServer) WriteResult(w http.ResponseWriter, result *messages.CartState) error {
+func (s *PoolServer) WriteResult(w http.ResponseWriter, result *CartGrain) error {
 	w.Header().Set("Content-Type", "application/json")
 	w.Header().Set("Cache-Control", "no-cache")
 	w.Header().Set("Access-Control-Allow-Origin", "*")
@@ -86,7 +70,7 @@ func (s *PoolServer) HandleDeleteItem(w http.ResponseWriter, r *http.Request, id
 	if err != nil {
 		return err
 	}
-	data, err := s.process(id, &messages.RemoveItem{Id: int64(itemId)})
+	data, err := s.ApplyLocal(id, &messages.RemoveItem{Id: int64(itemId)})
 	if err != nil {
 		return err
 	}
@@ -106,7 +90,7 @@ func (s *PoolServer) HandleSetDelivery(w http.ResponseWriter, r *http.Request, i
 	if err != nil {
 		return err
 	}
-	data, err := s.process(id, &messages.SetDelivery{
+	data, err := s.ApplyLocal(id, &messages.SetDelivery{
 		Provider:    delivery.Provider,
 		Items:       delivery.Items,
 		PickupPoint: delivery.PickupPoint,
@@ -129,7 +113,7 @@ func (s *PoolServer) HandleSetPickupPoint(w http.ResponseWriter, r *http.Request
 	if err != nil {
 		return err
 	}
-	reply, err := s.process(id, &messages.SetPickupPoint{
+	reply, err := s.ApplyLocal(id, &messages.SetPickupPoint{
 		DeliveryId: int64(deliveryId),
 		Id:         pickupPoint.Id,
 		Name:       pickupPoint.Name,
@@ -151,7 +135,7 @@ func (s *PoolServer) HandleRemoveDelivery(w http.ResponseWriter, r *http.Request
 	if err != nil {
 		return err
 	}
-	reply, err := s.process(id, &messages.RemoveDelivery{Id: int64(deliveryId)})
+	reply, err := s.ApplyLocal(id, &messages.RemoveDelivery{Id: int64(deliveryId)})
 	if err != nil {
 		return err
 	}
@@ -164,7 +148,7 @@ func (s *PoolServer) HandleQuantityChange(w http.ResponseWriter, r *http.Request
 	if err != nil {
 		return err
 	}
-	reply, err := s.process(id, &changeQuantity)
+	reply, err := s.ApplyLocal(id, &changeQuantity)
 	if err != nil {
 		return err
 	}
@@ -177,7 +161,7 @@ func (s *PoolServer) HandleSetCartItems(w http.ResponseWriter, r *http.Request,
 	if err != nil {
 		return err
 	}
-	reply, err := s.process(id, &setCartItems)
+	reply, err := s.ApplyLocal(id, &setCartItems)
 	if err != nil {
 		return err
 	}
@@ -190,7 +174,7 @@ func (s *PoolServer) HandleAddRequest(w http.ResponseWriter, r *http.Request, id
 	if err != nil {
 		return err
 	}
-	reply, err := s.process(id, &addRequest)
+	reply, err := s.ApplyLocal(id, &addRequest)
 	if err != nil {
 		return err
 	}
@@ -216,18 +200,35 @@ func (s *PoolServer) HandleConfirmation(w http.ResponseWriter, r *http.Request,
 	return json.NewEncoder(w).Encode(order)
 }
 
+func getCurrency(country string) string {
+	if country == "no" {
+		return "NOK"
+	}
+	return "SEK"
+}
+
+func getLocale(country string) string {
+	if country == "no" {
+		return "nb-no"
+	}
+	return "sv-se"
+}
+
 func (s *PoolServer) CreateOrUpdateCheckout(host string, id CartId) (*CheckoutOrder, error) {
+	country := getCountryFromHost(host)
 	meta := &CheckoutMeta{
 		Terms:        fmt.Sprintf("https://%s/terms", host),
 		Checkout:     fmt.Sprintf("https://%s/checkout?order_id={checkout.order.id}", host),
 		Confirmation: fmt.Sprintf("https://%s/confirmation/{checkout.order.id}", host),
 		Validation:   fmt.Sprintf("https://%s/validate", host),
 		Push:         fmt.Sprintf("https://%s/push?order_id={checkout.order.id}", host),
-		Country:      getCountryFromHost(host),
+		Country:      country,
+		Currency:     getCurrency(country),
+		Locale:       getLocale(country),
 	}
 
 	// Get current grain state (may be local or remote)
-	grain, err := s.pool.Get(id)
+	grain, err := s.pool.Get(uint64(id))
 	if err != nil {
 		return nil, err
 	}
@@ -247,7 +248,7 @@ func (s *PoolServer) CreateOrUpdateCheckout(host string, id CartId) (*CheckoutOr
 
 func (s *PoolServer) ApplyCheckoutStarted(klarnaOrder *CheckoutOrder, id CartId) (*CartGrain, error) {
 	// Persist initialization state via mutation (best-effort)
-	return s.pool.Apply(id, &messages.InitializeCheckout{
+	return s.pool.Apply(uint64(id), &messages.InitializeCheckout{
 		OrderId:           klarnaOrder.ID,
 		Status:            klarnaOrder.Status,
 		PaymentInProgress: true,
@@ -266,73 +267,58 @@ func (s *PoolServer) HandleCheckout(w http.ResponseWriter, r *http.Request, id C
 	return json.NewEncoder(w).Encode(klarnaOrder)
 }
 
-func NewCartId() CartId {
-	// Deprecated: legacy random/time based cart id generator.
-	// Retained for compatibility; new code should prefer canonical CartID path.
-	cid, err := NewCartID()
-	if err != nil {
-		// Fallback to legacy method only if crypto/rand fails
-		id := time.Now().UnixNano() + rand.Int63()
-		return ToCartId(fmt.Sprintf("%d", id))
+func CookieCartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		var id CartId
+		cookie, err := r.Cookie("cartid")
+		if err != nil || cookie.Value == "" {
+			id = MustNewCartId()
+			http.SetCookie(w, &http.Cookie{
+				Name:     "cartid",
+				Value:    id.String(),
+				Secure:   r.TLS != nil,
+				HttpOnly: true,
+				Path:     "/",
+				Expires:  time.Now().AddDate(0, 0, 14),
+				SameSite: http.SameSiteLaxMode,
+			})
+			w.Header().Set("Set-Cart-Id", id.String())
+		} else {
+			parsed, ok := ParseCartId(cookie.Value)
+			if !ok {
+				id = MustNewCartId()
+				http.SetCookie(w, &http.Cookie{
+					Name:     "cartid",
+					Value:    id.String(),
+					Secure:   r.TLS != nil,
+					HttpOnly: true,
+					Path:     "/",
+					Expires:  time.Now().AddDate(0, 0, 14),
+					SameSite: http.SameSiteLaxMode,
+				})
+				w.Header().Set("Set-Cart-Id", id.String())
+			} else {
+				id = parsed
+			}
+		}
+
+		err = fn(id, w, r)
+		if err != nil {
+			log.Printf("Server error, not remote error: %v\n", err)
+			w.WriteHeader(http.StatusInternalServerError)
+			w.Write([]byte(err.Error()))
+		}
+
 	}
-	return CartIDToLegacy(cid)
 }
 
-func CookieCartIdHandler(fn func(w http.ResponseWriter, r *http.Request, cartId CartId) error) func(w http.ResponseWriter, r *http.Request) error {
-	return func(w http.ResponseWriter, r *http.Request) error {
-		// Extract / normalize cookie (preserve legacy textual IDs without rewriting).
-		var legacy CartId
-		cookies := r.CookiesNamed("cartid")
-		if len(cookies) == 0 {
-			// No cookie -> generate new canonical base62 id.
-			cid, generated, _, err := CanonicalizeOrLegacy("")
-			if err != nil {
-				return fmt.Errorf("failed to generate cart id: %w", err)
-			}
-			legacy = CartIDToLegacy(cid)
-			if generated {
-				http.SetCookie(w, &http.Cookie{
-					Name:     "cartid",
-					Value:    cid.String(),
-					Secure:   r.TLS != nil,
-					HttpOnly: true,
-					Path:     "/",
-					Expires:  time.Now().AddDate(0, 0, 14),
-					SameSite: http.SameSiteLaxMode,
-				})
-				w.Header().Set("Set-Cart-Id", cid.String())
-			}
-		} else {
-			raw := cookies[0].Value
-			cid, generated, wasBase62, err := CanonicalizeOrLegacy(raw)
-			if err != nil {
-				return fmt.Errorf("failed to canonicalize cart id: %w", err)
-			}
-			legacy = CartIDToLegacy(cid)
-			// Only set a new cookie if we actually generated a brand-new ID (empty input).
-			// For legacy (non-base62) ids we preserve the original text and do not overwrite.
-			if generated && wasBase62 {
-				http.SetCookie(w, &http.Cookie{
-					Name:     "cartid",
-					Value:    cid.String(),
-					Secure:   r.TLS != nil,
-					HttpOnly: true,
-					Path:     "/",
-					Expires:  time.Now().AddDate(0, 0, 14),
-					SameSite: http.SameSiteLaxMode,
-				})
-				w.Header().Set("Set-Cart-Id", cid.String())
-			}
-		}
-		return fn(w, r, legacy)
-	}
-}
+// Removed leftover legacy block after CookieCartIdHandler (obsolete code referencing cid/legacy)
 
 func (s *PoolServer) RemoveCartCookie(w http.ResponseWriter, r *http.Request, cartId CartId) error {
-	cartId = NewCartId()
+	// Clear cart cookie (breaking change: do not issue a new legacy id here)
 	http.SetCookie(w, &http.Cookie{
 		Name:     "cartid",
-		Value:    cartId.String(),
+		Value:    "",
 		Path:     "/",
 		Secure:   r.TLS != nil,
 		HttpOnly: true,
@@ -343,26 +329,51 @@ func (s *PoolServer) RemoveCartCookie(w http.ResponseWriter, r *http.Request, ca
 	return nil
 }
 
-func CartIdHandler(fn func(w http.ResponseWriter, r *http.Request, cartId CartId) error) func(w http.ResponseWriter, r *http.Request) error {
-	return func(w http.ResponseWriter, r *http.Request) error {
+func CartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		var id CartId
 		raw := r.PathValue("id")
-		cid, generated, wasBase62, err := CanonicalizeOrLegacy(raw)
+		// If no id supplied, generate a new one
+		if raw == "" {
+			id := MustNewCartId()
+			w.Header().Set("Set-Cart-Id", id.String())
+		} else {
+			// Parse base62 cart id
+			if parsedId, ok := ParseCartId(raw); !ok {
+				w.WriteHeader(http.StatusBadRequest)
+				w.Write([]byte("cart id is invalid"))
+				return
+			} else {
+				id = parsedId
+			}
+		}
+
+		err := fn(id, w, r)
 		if err != nil {
-			return fmt.Errorf("invalid cart id: %w", err)
+			log.Printf("Server error, not remote error: %v\n", err)
+			w.WriteHeader(http.StatusInternalServerError)
+			w.Write([]byte(err.Error()))
 		}
-		legacy := CartIDToLegacy(cid)
-		// Only emit Set-Cart-Id header if we produced a brand-new canonical id
-		// AND it is base62 (avoid rewriting legacy textual identifiers).
-		if generated && wasBase62 {
-			w.Header().Set("Set-Cart-Id", cid.String())
 	}
-		return fn(w, r, legacy)
+}
+
+func (s *PoolServer) ProxyHandler(fn func(w http.ResponseWriter, r *http.Request, cartId CartId) error) func(cartId CartId, w http.ResponseWriter, r *http.Request) error {
+	return func(cartId CartId, w http.ResponseWriter, r *http.Request) error {
+		if ownerHost, ok := s.pool.OwnerHost(uint64(cartId)); ok {
+			handled, err := ownerHost.Proxy(uint64(cartId), w, r)
+			if err == nil && handled {
+				return nil
+			}
+		}
+
+		return fn(w, r, cartId)
+
 	}
 }
 
 func (s *PoolServer) Serve() *http.ServeMux {
+
 	mux := http.NewServeMux()
-	//mux.HandleFunc("/", s.RewritePath)
 	mux.HandleFunc("OPTIONS /", func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Access-Control-Allow-Origin", "*")
 		w.Header().Set("Access-Control-Allow-Methods", "GET, PUT, POST, DELETE")
@@ -370,29 +381,29 @@ func (s *PoolServer) Serve() *http.ServeMux {
 		w.WriteHeader(http.StatusOK)
 	})
 
-	mux.HandleFunc("GET /", ErrorHandler(CookieCartIdHandler(s.HandleGet)))
-	mux.HandleFunc("GET /add/{sku}", ErrorHandler(CookieCartIdHandler(s.HandleAddSku)))
-	mux.HandleFunc("POST /", ErrorHandler(CookieCartIdHandler(s.HandleAddRequest)))
-	mux.HandleFunc("POST /set", ErrorHandler(CookieCartIdHandler(s.HandleSetCartItems)))
-	mux.HandleFunc("DELETE /{itemId}", ErrorHandler(CookieCartIdHandler(s.HandleDeleteItem)))
-	mux.HandleFunc("PUT /", ErrorHandler(CookieCartIdHandler(s.HandleQuantityChange)))
-	mux.HandleFunc("DELETE /", ErrorHandler(CookieCartIdHandler(s.RemoveCartCookie)))
-	mux.HandleFunc("POST /delivery", ErrorHandler(CookieCartIdHandler(s.HandleSetDelivery)))
-	mux.HandleFunc("DELETE /delivery/{deliveryId}", ErrorHandler(CookieCartIdHandler(s.HandleRemoveDelivery)))
-	mux.HandleFunc("PUT /delivery/{deliveryId}/pickupPoint", ErrorHandler(CookieCartIdHandler(s.HandleSetPickupPoint)))
-	mux.HandleFunc("GET /checkout", ErrorHandler(CookieCartIdHandler(s.HandleCheckout)))
-	mux.HandleFunc("GET /confirmation/{orderId}", ErrorHandler(CookieCartIdHandler(s.HandleConfirmation)))
+	mux.HandleFunc("GET /", CookieCartIdHandler(s.ProxyHandler(s.HandleGet)))
+	mux.HandleFunc("GET /add/{sku}", CookieCartIdHandler(s.ProxyHandler(s.HandleAddSku)))
+	mux.HandleFunc("POST /", CookieCartIdHandler(s.ProxyHandler(s.HandleAddRequest)))
+	mux.HandleFunc("POST /set", CookieCartIdHandler(s.ProxyHandler(s.HandleSetCartItems)))
+	mux.HandleFunc("DELETE /{itemId}", CookieCartIdHandler(s.ProxyHandler(s.HandleDeleteItem)))
+	mux.HandleFunc("PUT /", CookieCartIdHandler(s.ProxyHandler(s.HandleQuantityChange)))
+	mux.HandleFunc("DELETE /", CookieCartIdHandler(s.ProxyHandler(s.RemoveCartCookie)))
+	mux.HandleFunc("POST /delivery", CookieCartIdHandler(s.ProxyHandler(s.HandleSetDelivery)))
+	mux.HandleFunc("DELETE /delivery/{deliveryId}", CookieCartIdHandler(s.ProxyHandler(s.HandleRemoveDelivery)))
+	mux.HandleFunc("PUT /delivery/{deliveryId}/pickupPoint", CookieCartIdHandler(s.ProxyHandler(s.HandleSetPickupPoint)))
+	mux.HandleFunc("GET /checkout", CookieCartIdHandler(s.ProxyHandler(s.HandleCheckout)))
+	mux.HandleFunc("GET /confirmation/{orderId}", CookieCartIdHandler(s.ProxyHandler(s.HandleConfirmation)))
 
-	mux.HandleFunc("GET /byid/{id}", ErrorHandler(CartIdHandler(s.HandleGet)))
-	mux.HandleFunc("GET /byid/{id}/add/{sku}", ErrorHandler(CartIdHandler(s.HandleAddSku)))
-	mux.HandleFunc("POST /byid/{id}", ErrorHandler(CartIdHandler(s.HandleAddRequest)))
-	mux.HandleFunc("DELETE /byid/{id}/{itemId}", ErrorHandler(CartIdHandler(s.HandleDeleteItem)))
-	mux.HandleFunc("PUT /byid/{id}", ErrorHandler(CartIdHandler(s.HandleQuantityChange)))
-	mux.HandleFunc("POST /byid/{id}/delivery", ErrorHandler(CartIdHandler(s.HandleSetDelivery)))
-	mux.HandleFunc("DELETE /byid/{id}/delivery/{deliveryId}", ErrorHandler(CartIdHandler(s.HandleRemoveDelivery)))
-	mux.HandleFunc("PUT /byid/{id}/delivery/{deliveryId}/pickupPoint", ErrorHandler(CartIdHandler(s.HandleSetPickupPoint)))
-	mux.HandleFunc("GET /byid/{id}/checkout", ErrorHandler(CartIdHandler(s.HandleCheckout)))
-	mux.HandleFunc("GET /byid/{id}/confirmation", ErrorHandler(CartIdHandler(s.HandleConfirmation)))
+	mux.HandleFunc("GET /byid/{id}", CartIdHandler(s.ProxyHandler(s.HandleGet)))
+	mux.HandleFunc("GET /byid/{id}/add/{sku}", CartIdHandler(s.ProxyHandler(s.HandleAddSku)))
+	mux.HandleFunc("POST /byid/{id}", CartIdHandler(s.ProxyHandler(s.HandleAddRequest)))
+	mux.HandleFunc("DELETE /byid/{id}/{itemId}", CartIdHandler(s.ProxyHandler(s.HandleDeleteItem)))
+	mux.HandleFunc("PUT /byid/{id}", CartIdHandler(s.ProxyHandler(s.HandleQuantityChange)))
+	mux.HandleFunc("POST /byid/{id}/delivery", CartIdHandler(s.ProxyHandler(s.HandleSetDelivery)))
+	mux.HandleFunc("DELETE /byid/{id}/delivery/{deliveryId}", CartIdHandler(s.ProxyHandler(s.HandleRemoveDelivery)))
+	mux.HandleFunc("PUT /byid/{id}/delivery/{deliveryId}/pickupPoint", CartIdHandler(s.ProxyHandler(s.HandleSetPickupPoint)))
+	mux.HandleFunc("GET /byid/{id}/checkout", CartIdHandler(s.ProxyHandler(s.HandleCheckout)))
+	mux.HandleFunc("GET /byid/{id}/confirmation", CartIdHandler(s.ProxyHandler(s.HandleConfirmation)))
 
 	return mux
 }

proto/cart_actor.proto

@@ -1,187 +0,0 @@
-syntax = "proto3";
-
-package messages;
-
-option go_package = "git.tornberg.me/go-cart-actor/proto;messages";
-
-import "messages.proto";
-
-// -----------------------------------------------------------------------------
-// Cart Actor gRPC API (Breaking v2 - Per-Mutation RPCs)
-// -----------------------------------------------------------------------------
-// This version removes the previous MutationEnvelope + Mutate RPC.
-// Each mutation now has its own request wrapper and dedicated RPC method
-// providing simpler, type-focused client stubs and enabling per-mutation
-// metrics, auth and rate limiting.
-//
-// Regenerate Go code after editing:
-//   protoc --go_out=. --go_opt=paths=source_relative \
-//          --go-grpc_out=. --go-grpc_opt=paths=source_relative \
-//          proto/cart_actor.proto proto/messages.proto
-//
-// Backward compatibility: This is a breaking change (old clients must update).
-// -----------------------------------------------------------------------------
-
-// Shared reply for all mutation RPCs.
-message CartMutationReply {
-  int32 status_code = 1;          // HTTP-like status (200 success, 4xx client, 5xx server)
-  oneof result {
-    CartState state = 2;          // Updated cart state on success
-    string error = 3;             // Error message on failure
-  }
-  int64 server_timestamp = 4;     // Server-assigned Unix timestamp (optional auditing)
-}
-
-// Fetch current cart state without mutation.
-message StateRequest {
-  string cart_id = 1;
-}
-
-message StateReply {
-  int32 status_code = 1;
-  oneof result {
-    CartState state = 2;
-    string error = 3;
-  }
-}
-
-// Per-mutation request wrappers. We wrap the existing inner mutation
-// messages (defined in messages.proto) to add cart_id + optional metadata
-// without altering the inner message definitions.
-
-message AddRequestRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  AddRequest payload = 10;
-}
-
-message AddItemRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  AddItem payload = 10;
-}
-
-message RemoveItemRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  RemoveItem payload = 10;
-}
-
-message RemoveDeliveryRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  RemoveDelivery payload = 10;
-}
-
-message ChangeQuantityRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  ChangeQuantity payload = 10;
-}
-
-message SetDeliveryRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  SetDelivery payload = 10;
-}
-
-message SetPickupPointRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  SetPickupPoint payload = 10;
-}
-
-message CreateCheckoutOrderRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  CreateCheckoutOrder payload = 10;
-}
-
-message SetCartItemsRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  SetCartRequest payload = 10;
-}
-
-message OrderCompletedRequest {
-  string cart_id = 1;
-  int64 client_timestamp = 2;
-  OrderCreated payload = 10;
-}
-
-// Excerpt: updated messages for camelCase JSON output
-message CartState {
-  string id = 1;                         // was cart_id
-  repeated CartItemState items = 2;
-  int64 totalPrice = 3;                  // was total_price
-  int64 totalTax = 4;                    // was total_tax
-  int64 totalDiscount = 5;               // was total_discount
-  repeated DeliveryState deliveries = 6;
-  bool paymentInProgress = 7;            // was payment_in_progress
-  string orderReference = 8;             // was order_reference
-  string paymentStatus = 9;              // was payment_status
-  bool processing = 10;                  // NEW (mirrors legacy CartGrain.processing)
-}
-
-message CartItemState {
-  int64 id = 1;
-  int64 itemId = 2;                      // was source_item_id
-  string sku = 3;
-  string name = 4;
-  int64 price = 5;                       // was unit_price
-  int32 qty = 6;                         // was quantity
-  int64 totalPrice = 7;                  // was total_price
-  int64 totalTax = 8;                    // was total_tax
-  int64 orgPrice = 9;                    // was org_price
-  int32 taxRate = 10;                    // was tax_rate
-  int64 totalDiscount = 11;
-  string brand = 12;
-  string category = 13;
-  string category2 = 14;
-  string category3 = 15;
-  string category4 = 16;
-  string category5 = 17;
-  string image = 18;
-  string type = 19;                      // was article_type
-  string sellerId = 20;                  // was seller_id
-  string sellerName = 21;                // was seller_name
-  string disclaimer = 22;
-  string outlet = 23;
-  string storeId = 24;                   // was store_id
-  int32 stock = 25;
-}
-
-message DeliveryState {
-  int64 id = 1;
-  string provider = 2;
-  int64 price = 3;
-  repeated int64 items = 4;              // was item_ids
-  PickupPoint pickupPoint = 5;           // was pickup_point
-}
-
-// (CheckoutRequest / CheckoutReply removed - checkout handled at HTTP layer)
-
-// -----------------------------------------------------------------------------
-// Service definition (per-mutation RPCs + checkout)
-// -----------------------------------------------------------------------------
-service CartActor {
-  rpc AddRequest(AddRequestRequest) returns (CartMutationReply);
-  rpc AddItem(AddItemRequest) returns (CartMutationReply);
-  rpc RemoveItem(RemoveItemRequest) returns (CartMutationReply);
-  rpc RemoveDelivery(RemoveDeliveryRequest) returns (CartMutationReply);
-  rpc ChangeQuantity(ChangeQuantityRequest) returns (CartMutationReply);
-  rpc SetDelivery(SetDeliveryRequest) returns (CartMutationReply);
-  rpc SetPickupPoint(SetPickupPointRequest) returns (CartMutationReply);
-  // (Checkout RPC removed - handled externally)
-  rpc SetCartItems(SetCartItemsRequest) returns (CartMutationReply);
-  rpc OrderCompleted(OrderCompletedRequest) returns (CartMutationReply);
-
-  rpc GetState(StateRequest) returns (StateReply);
-}
-
-// -----------------------------------------------------------------------------
-// Future enhancements:
-// * BatchMutate RPC (repeated heterogeneous mutations)
-// * Streaming state updates (WatchState)
-// * Versioning / optimistic concurrency control
-// -----------------------------------------------------------------------------

proto/cart_actor_grpc.pb.go

@@ -1,473 +0,0 @@
-// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
-// versions:
-// - protoc-gen-go-grpc v1.5.1
-// - protoc             v3.21.12
-// source: cart_actor.proto
-
-package 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 (
-	CartActor_AddRequest_FullMethodName     = "/messages.CartActor/AddRequest"
-	CartActor_AddItem_FullMethodName        = "/messages.CartActor/AddItem"
-	CartActor_RemoveItem_FullMethodName     = "/messages.CartActor/RemoveItem"
-	CartActor_RemoveDelivery_FullMethodName = "/messages.CartActor/RemoveDelivery"
-	CartActor_ChangeQuantity_FullMethodName = "/messages.CartActor/ChangeQuantity"
-	CartActor_SetDelivery_FullMethodName    = "/messages.CartActor/SetDelivery"
-	CartActor_SetPickupPoint_FullMethodName = "/messages.CartActor/SetPickupPoint"
-	CartActor_SetCartItems_FullMethodName   = "/messages.CartActor/SetCartItems"
-	CartActor_OrderCompleted_FullMethodName = "/messages.CartActor/OrderCompleted"
-	CartActor_GetState_FullMethodName       = "/messages.CartActor/GetState"
-)
-
-// CartActorClient is the client API for CartActor 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.
-//
-// -----------------------------------------------------------------------------
-// Service definition (per-mutation RPCs + checkout)
-// -----------------------------------------------------------------------------
-type CartActorClient interface {
-	AddRequest(ctx context.Context, in *AddRequestRequest, opts ...grpc.CallOption) (*CartMutationReply, error)
-	AddItem(ctx context.Context, in *AddItemRequest, opts ...grpc.CallOption) (*CartMutationReply, error)
-	RemoveItem(ctx context.Context, in *RemoveItemRequest, opts ...grpc.CallOption) (*CartMutationReply, error)
-	RemoveDelivery(ctx context.Context, in *RemoveDeliveryRequest, opts ...grpc.CallOption) (*CartMutationReply, error)
-	ChangeQuantity(ctx context.Context, in *ChangeQuantityRequest, opts ...grpc.CallOption) (*CartMutationReply, error)
-	SetDelivery(ctx context.Context, in *SetDeliveryRequest, opts ...grpc.CallOption) (*CartMutationReply, error)
-	SetPickupPoint(ctx context.Context, in *SetPickupPointRequest, opts ...grpc.CallOption) (*CartMutationReply, error)
-	// (Checkout RPC removed - handled externally)
-	SetCartItems(ctx context.Context, in *SetCartItemsRequest, opts ...grpc.CallOption) (*CartMutationReply, error)
-	OrderCompleted(ctx context.Context, in *OrderCompletedRequest, opts ...grpc.CallOption) (*CartMutationReply, error)
-	GetState(ctx context.Context, in *StateRequest, opts ...grpc.CallOption) (*StateReply, error)
-}
-
-type cartActorClient struct {
-	cc grpc.ClientConnInterface
-}
-
-func NewCartActorClient(cc grpc.ClientConnInterface) CartActorClient {
-	return &cartActorClient{cc}
-}
-
-func (c *cartActorClient) AddRequest(ctx context.Context, in *AddRequestRequest, opts ...grpc.CallOption) (*CartMutationReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartMutationReply)
-	err := c.cc.Invoke(ctx, CartActor_AddRequest_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *cartActorClient) AddItem(ctx context.Context, in *AddItemRequest, opts ...grpc.CallOption) (*CartMutationReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartMutationReply)
-	err := c.cc.Invoke(ctx, CartActor_AddItem_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *cartActorClient) RemoveItem(ctx context.Context, in *RemoveItemRequest, opts ...grpc.CallOption) (*CartMutationReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartMutationReply)
-	err := c.cc.Invoke(ctx, CartActor_RemoveItem_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *cartActorClient) RemoveDelivery(ctx context.Context, in *RemoveDeliveryRequest, opts ...grpc.CallOption) (*CartMutationReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartMutationReply)
-	err := c.cc.Invoke(ctx, CartActor_RemoveDelivery_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *cartActorClient) ChangeQuantity(ctx context.Context, in *ChangeQuantityRequest, opts ...grpc.CallOption) (*CartMutationReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartMutationReply)
-	err := c.cc.Invoke(ctx, CartActor_ChangeQuantity_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *cartActorClient) SetDelivery(ctx context.Context, in *SetDeliveryRequest, opts ...grpc.CallOption) (*CartMutationReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartMutationReply)
-	err := c.cc.Invoke(ctx, CartActor_SetDelivery_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *cartActorClient) SetPickupPoint(ctx context.Context, in *SetPickupPointRequest, opts ...grpc.CallOption) (*CartMutationReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartMutationReply)
-	err := c.cc.Invoke(ctx, CartActor_SetPickupPoint_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *cartActorClient) SetCartItems(ctx context.Context, in *SetCartItemsRequest, opts ...grpc.CallOption) (*CartMutationReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartMutationReply)
-	err := c.cc.Invoke(ctx, CartActor_SetCartItems_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *cartActorClient) OrderCompleted(ctx context.Context, in *OrderCompletedRequest, opts ...grpc.CallOption) (*CartMutationReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(CartMutationReply)
-	err := c.cc.Invoke(ctx, CartActor_OrderCompleted_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *cartActorClient) GetState(ctx context.Context, in *StateRequest, opts ...grpc.CallOption) (*StateReply, error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	out := new(StateReply)
-	err := c.cc.Invoke(ctx, CartActor_GetState_FullMethodName, in, out, cOpts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-// CartActorServer is the server API for CartActor service.
-// All implementations must embed UnimplementedCartActorServer
-// for forward compatibility.
-//
-// -----------------------------------------------------------------------------
-// Service definition (per-mutation RPCs + checkout)
-// -----------------------------------------------------------------------------
-type CartActorServer interface {
-	AddRequest(context.Context, *AddRequestRequest) (*CartMutationReply, error)
-	AddItem(context.Context, *AddItemRequest) (*CartMutationReply, error)
-	RemoveItem(context.Context, *RemoveItemRequest) (*CartMutationReply, error)
-	RemoveDelivery(context.Context, *RemoveDeliveryRequest) (*CartMutationReply, error)
-	ChangeQuantity(context.Context, *ChangeQuantityRequest) (*CartMutationReply, error)
-	SetDelivery(context.Context, *SetDeliveryRequest) (*CartMutationReply, error)
-	SetPickupPoint(context.Context, *SetPickupPointRequest) (*CartMutationReply, error)
-	// (Checkout RPC removed - handled externally)
-	SetCartItems(context.Context, *SetCartItemsRequest) (*CartMutationReply, error)
-	OrderCompleted(context.Context, *OrderCompletedRequest) (*CartMutationReply, error)
-	GetState(context.Context, *StateRequest) (*StateReply, error)
-	mustEmbedUnimplementedCartActorServer()
-}
-
-// UnimplementedCartActorServer 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 UnimplementedCartActorServer struct{}
-
-func (UnimplementedCartActorServer) AddRequest(context.Context, *AddRequestRequest) (*CartMutationReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method AddRequest not implemented")
-}
-func (UnimplementedCartActorServer) AddItem(context.Context, *AddItemRequest) (*CartMutationReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method AddItem not implemented")
-}
-func (UnimplementedCartActorServer) RemoveItem(context.Context, *RemoveItemRequest) (*CartMutationReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method RemoveItem not implemented")
-}
-func (UnimplementedCartActorServer) RemoveDelivery(context.Context, *RemoveDeliveryRequest) (*CartMutationReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method RemoveDelivery not implemented")
-}
-func (UnimplementedCartActorServer) ChangeQuantity(context.Context, *ChangeQuantityRequest) (*CartMutationReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method ChangeQuantity not implemented")
-}
-func (UnimplementedCartActorServer) SetDelivery(context.Context, *SetDeliveryRequest) (*CartMutationReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method SetDelivery not implemented")
-}
-func (UnimplementedCartActorServer) SetPickupPoint(context.Context, *SetPickupPointRequest) (*CartMutationReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method SetPickupPoint not implemented")
-}
-func (UnimplementedCartActorServer) SetCartItems(context.Context, *SetCartItemsRequest) (*CartMutationReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method SetCartItems not implemented")
-}
-func (UnimplementedCartActorServer) OrderCompleted(context.Context, *OrderCompletedRequest) (*CartMutationReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method OrderCompleted not implemented")
-}
-func (UnimplementedCartActorServer) GetState(context.Context, *StateRequest) (*StateReply, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method GetState not implemented")
-}
-func (UnimplementedCartActorServer) mustEmbedUnimplementedCartActorServer() {}
-func (UnimplementedCartActorServer) testEmbeddedByValue()                   {}
-
-// UnsafeCartActorServer may be embedded to opt out of forward compatibility for this service.
-// Use of this interface is not recommended, as added methods to CartActorServer will
-// result in compilation errors.
-type UnsafeCartActorServer interface {
-	mustEmbedUnimplementedCartActorServer()
-}
-
-func RegisterCartActorServer(s grpc.ServiceRegistrar, srv CartActorServer) {
-	// If the following call pancis, it indicates UnimplementedCartActorServer 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(&CartActor_ServiceDesc, srv)
-}
-
-func _CartActor_AddRequest_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(AddRequestRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).AddRequest(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_AddRequest_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).AddRequest(ctx, req.(*AddRequestRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _CartActor_AddItem_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(AddItemRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).AddItem(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_AddItem_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).AddItem(ctx, req.(*AddItemRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _CartActor_RemoveItem_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(RemoveItemRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).RemoveItem(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_RemoveItem_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).RemoveItem(ctx, req.(*RemoveItemRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _CartActor_RemoveDelivery_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(RemoveDeliveryRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).RemoveDelivery(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_RemoveDelivery_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).RemoveDelivery(ctx, req.(*RemoveDeliveryRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _CartActor_ChangeQuantity_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(ChangeQuantityRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).ChangeQuantity(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_ChangeQuantity_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).ChangeQuantity(ctx, req.(*ChangeQuantityRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _CartActor_SetDelivery_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(SetDeliveryRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).SetDelivery(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_SetDelivery_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).SetDelivery(ctx, req.(*SetDeliveryRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _CartActor_SetPickupPoint_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(SetPickupPointRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).SetPickupPoint(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_SetPickupPoint_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).SetPickupPoint(ctx, req.(*SetPickupPointRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _CartActor_SetCartItems_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(SetCartItemsRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).SetCartItems(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_SetCartItems_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).SetCartItems(ctx, req.(*SetCartItemsRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _CartActor_OrderCompleted_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(OrderCompletedRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).OrderCompleted(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_OrderCompleted_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).OrderCompleted(ctx, req.(*OrderCompletedRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _CartActor_GetState_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(StateRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(CartActorServer).GetState(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: CartActor_GetState_FullMethodName,
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(CartActorServer).GetState(ctx, req.(*StateRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-// CartActor_ServiceDesc is the grpc.ServiceDesc for CartActor service.
-// It's only intended for direct use with grpc.RegisterService,
-// and not to be introspected or modified (even as a copy)
-var CartActor_ServiceDesc = grpc.ServiceDesc{
-	ServiceName: "messages.CartActor",
-	HandlerType: (*CartActorServer)(nil),
-	Methods: []grpc.MethodDesc{
-		{
-			MethodName: "AddRequest",
-			Handler:    _CartActor_AddRequest_Handler,
-		},
-		{
-			MethodName: "AddItem",
-			Handler:    _CartActor_AddItem_Handler,
-		},
-		{
-			MethodName: "RemoveItem",
-			Handler:    _CartActor_RemoveItem_Handler,
-		},
-		{
-			MethodName: "RemoveDelivery",
-			Handler:    _CartActor_RemoveDelivery_Handler,
-		},
-		{
-			MethodName: "ChangeQuantity",
-			Handler:    _CartActor_ChangeQuantity_Handler,
-		},
-		{
-			MethodName: "SetDelivery",
-			Handler:    _CartActor_SetDelivery_Handler,
-		},
-		{
-			MethodName: "SetPickupPoint",
-			Handler:    _CartActor_SetPickupPoint_Handler,
-		},
-		{
-			MethodName: "SetCartItems",
-			Handler:    _CartActor_SetCartItems_Handler,
-		},
-		{
-			MethodName: "OrderCompleted",
-			Handler:    _CartActor_OrderCompleted_Handler,
-		},
-		{
-			MethodName: "GetState",
-			Handler:    _CartActor_GetState_Handler,
-		},
-	},
-	Streams:  []grpc.StreamDesc{},
-	Metadata: "cart_actor.proto",
-}

proto/control_plane.proto

@@ -37,8 +37,8 @@ message NegotiateReply {
 }
 
 // CartIdsReply returns the list of cart IDs (string form) currently owned locally.
-message CartIdsReply {
-  repeated string cart_ids = 1;
+message ActorIdsReply {
+  repeated uint64 ids = 1;
 }
 
 // OwnerChangeAck retained as response type for Closing RPC (ConfirmOwner removed).
@@ -52,6 +52,19 @@ message ClosingNotice {
   string host = 1;
 }
 
+// OwnershipAnnounce broadcasts first-touch ownership claims for cart IDs.
+// First claim wins; receivers SHOULD NOT overwrite an existing different owner.
+message OwnershipAnnounce {
+  string host = 1;               // announcing host
+  repeated uint64 ids = 2;   // newly claimed cart ids
+}
+
+// ExpiryAnnounce broadcasts that a host evicted the provided cart IDs.
+message ExpiryAnnounce {
+  string host = 1;
+  repeated uint64 ids = 2;
+}
+
 // ControlPlane defines cluster coordination and ownership operations.
 service ControlPlane {
   // Ping for liveness; lightweight health signal.
@@ -61,10 +74,16 @@ service ControlPlane {
   rpc Negotiate(NegotiateRequest) returns (NegotiateReply);
 
   // GetCartIds lists currently owned cart IDs on this node.
-  rpc GetCartIds(Empty) returns (CartIdsReply);
+  rpc GetLocalActorIds(Empty) returns (ActorIdsReply);
 
   // ConfirmOwner RPC removed (was legacy ownership acknowledgement; ring-based ownership now authoritative)
 
+  // Ownership announcement: first-touch claim broadcast (idempotent; best-effort).
+  rpc AnnounceOwnership(OwnershipAnnounce) returns (OwnerChangeAck);
+
+  // Expiry announcement: drop remote ownership hints when local TTL expires.
+  rpc AnnounceExpiry(ExpiryAnnounce) returns (OwnerChangeAck);
+
   // Closing announces graceful shutdown so peers can proactively adjust.
   rpc Closing(ClosingNotice) returns (OwnerChangeAck);
 }

remote_grain_grpc.go

@@ -1,341 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"time"
-
-	proto "git.tornberg.me/go-cart-actor/proto" // generated package name is 'messages'; aliased as proto for consistency
-	"google.golang.org/grpc"
-)
-
-// RemoteGrainGRPC is the gRPC-backed implementation of a remote grain.
-// It mirrors the previous RemoteGrain (TCP/frame based) while using the
-// new CartActor gRPC service. It implements the Grain interface so that
-// SyncedPool can remain largely unchanged when swapping transport layers.
-type RemoteGrainGRPC struct {
-	Id     CartId
-	Host   string
-	client proto.CartActorClient
-	// Optional: keep the underlying conn so higher-level code can close if needed
-	conn *grpc.ClientConn
-
-	// Per-call timeout settings (tunable)
-	mutateTimeout time.Duration
-	stateTimeout  time.Duration
-}
-
-// NewRemoteGrainGRPC constructs a remote grain adapter from an existing gRPC client.
-func NewRemoteGrainGRPC(id CartId, host string, client proto.CartActorClient) *RemoteGrainGRPC {
-	return &RemoteGrainGRPC{
-		Id:            id,
-		Host:          host,
-		client:        client,
-		mutateTimeout: 800 * time.Millisecond,
-		stateTimeout:  400 * time.Millisecond,
-	}
-}
-
-// NewRemoteGrainGRPCWithConn dials the target and creates the gRPC client.
-// target should be host:port (where the CartActor service is exposed).
-func NewRemoteGrainGRPCWithConn(id CartId, host string, target string, dialOpts ...grpc.DialOption) (*RemoteGrainGRPC, error) {
-	// NOTE: insecure for initial migration; should be replaced with TLS later.
-	baseOpts := []grpc.DialOption{grpc.WithInsecure(), grpc.WithBlock()}
-	baseOpts = append(baseOpts, dialOpts...)
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	conn, err := grpc.DialContext(ctx, target, baseOpts...)
-	if err != nil {
-		return nil, err
-	}
-
-	client := proto.NewCartActorClient(conn)
-	return &RemoteGrainGRPC{
-		Id:            id,
-		Host:          host,
-		client:        client,
-		conn:          conn,
-		mutateTimeout: 800 * time.Millisecond,
-		stateTimeout:  400 * time.Millisecond,
-	}, nil
-}
-
-func (g *RemoteGrainGRPC) GetId() CartId {
-	return g.Id
-}
-
-// Apply executes a cart mutation via per-mutation RPCs (breaking v2 API)
-// and returns a *CartGrain reconstructed from the CartMutationReply state.
-func (g *RemoteGrainGRPC) Apply(content interface{}, isReplay bool) (*CartGrain, error) {
-	if isReplay {
-		return nil, fmt.Errorf("replay not supported for remote grains")
-	}
-	if content == nil {
-		return nil, fmt.Errorf("nil mutation content")
-	}
-
-	ts := time.Now().Unix()
-
-	var invoke func(ctx context.Context) (*proto.CartMutationReply, error)
-
-	switch m := content.(type) {
-	case *proto.AddRequest:
-		invoke = func(ctx context.Context) (*proto.CartMutationReply, error) {
-			return g.client.AddRequest(ctx, &proto.AddRequestRequest{
-				CartId:          g.Id.String(),
-				ClientTimestamp: ts,
-				Payload:         m,
-			})
-		}
-	case *proto.AddItem:
-		invoke = func(ctx context.Context) (*proto.CartMutationReply, error) {
-			return g.client.AddItem(ctx, &proto.AddItemRequest{
-				CartId:          g.Id.String(),
-				ClientTimestamp: ts,
-				Payload:         m,
-			})
-		}
-	case *proto.RemoveItem:
-		invoke = func(ctx context.Context) (*proto.CartMutationReply, error) {
-			return g.client.RemoveItem(ctx, &proto.RemoveItemRequest{
-				CartId:          g.Id.String(),
-				ClientTimestamp: ts,
-				Payload:         m,
-			})
-		}
-	case *proto.RemoveDelivery:
-		invoke = func(ctx context.Context) (*proto.CartMutationReply, error) {
-			return g.client.RemoveDelivery(ctx, &proto.RemoveDeliveryRequest{
-				CartId:          g.Id.String(),
-				ClientTimestamp: ts,
-				Payload:         m,
-			})
-		}
-	case *proto.ChangeQuantity:
-		invoke = func(ctx context.Context) (*proto.CartMutationReply, error) {
-			return g.client.ChangeQuantity(ctx, &proto.ChangeQuantityRequest{
-				CartId:          g.Id.String(),
-				ClientTimestamp: ts,
-				Payload:         m,
-			})
-		}
-	case *proto.SetDelivery:
-		invoke = func(ctx context.Context) (*proto.CartMutationReply, error) {
-			return g.client.SetDelivery(ctx, &proto.SetDeliveryRequest{
-				CartId:          g.Id.String(),
-				ClientTimestamp: ts,
-				Payload:         m,
-			})
-		}
-	case *proto.SetPickupPoint:
-		invoke = func(ctx context.Context) (*proto.CartMutationReply, error) {
-			return g.client.SetPickupPoint(ctx, &proto.SetPickupPointRequest{
-				CartId:          g.Id.String(),
-				ClientTimestamp: ts,
-				Payload:         m,
-			})
-		}
-	case *proto.CreateCheckoutOrder:
-		return nil, fmt.Errorf("CreateCheckoutOrder deprecated: checkout is handled via HTTP endpoint (HandleCheckout)")
-	case *proto.SetCartRequest:
-		invoke = func(ctx context.Context) (*proto.CartMutationReply, error) {
-			return g.client.SetCartItems(ctx, &proto.SetCartItemsRequest{
-				CartId:          g.Id.String(),
-				ClientTimestamp: ts,
-				Payload:         m,
-			})
-		}
-	case *proto.OrderCreated:
-		invoke = func(ctx context.Context) (*proto.CartMutationReply, error) {
-			return g.client.OrderCompleted(ctx, &proto.OrderCompletedRequest{
-				CartId:          g.Id.String(),
-				ClientTimestamp: ts,
-				Payload:         m,
-			})
-		}
-	default:
-		return nil, fmt.Errorf("unsupported mutation type %T", content)
-	}
-
-	if invoke == nil {
-		return nil, fmt.Errorf("no invocation mapped for mutation %T", content)
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), g.mutateTimeout)
-	defer cancel()
-
-	resp, err := invoke(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		if e := resp.GetError(); e != "" {
-			return nil, fmt.Errorf("remote mutation failed %d: %s", resp.StatusCode, e)
-		}
-		return nil, fmt.Errorf("remote mutation failed %d", resp.StatusCode)
-	}
-	state := resp.GetState()
-	if state == nil {
-		return nil, fmt.Errorf("mutation reply missing state on success")
-	}
-	// Reconstruct a lightweight CartGrain (only fields we expose internally)
-	grain := &CartGrain{
-		Id:                ToCartId(state.Id),
-		TotalPrice:        state.TotalPrice,
-		TotalTax:          state.TotalTax,
-		TotalDiscount:     state.TotalDiscount,
-		PaymentInProgress: state.PaymentInProgress,
-		OrderReference:    state.OrderReference,
-		PaymentStatus:     state.PaymentStatus,
-	}
-	// Items
-	for _, it := range state.Items {
-		if it == nil {
-			continue
-		}
-		outlet := toPtr(it.Outlet)
-		storeId := toPtr(it.StoreId)
-		grain.Items = append(grain.Items, &CartItem{
-			Id:          int(it.Id),
-			ItemId:      int(it.ItemId),
-			Sku:         it.Sku,
-			Name:        it.Name,
-			Price:       it.Price,
-			Quantity:    int(it.Qty),
-			TotalPrice:  it.TotalPrice,
-			TotalTax:    it.TotalTax,
-			OrgPrice:    it.OrgPrice,
-			TaxRate:     int(it.TaxRate),
-			Brand:       it.Brand,
-			Category:    it.Category,
-			Category2:   it.Category2,
-			Category3:   it.Category3,
-			Category4:   it.Category4,
-			Category5:   it.Category5,
-			Image:       it.Image,
-			ArticleType: it.Type,
-			SellerId:    it.SellerId,
-			SellerName:  it.SellerName,
-			Disclaimer:  it.Disclaimer,
-			Outlet:      outlet,
-			StoreId:     storeId,
-			Stock:       StockStatus(it.Stock),
-		})
-	}
-	// Deliveries
-	for _, d := range state.Deliveries {
-		if d == nil {
-			continue
-		}
-		intIds := make([]int, 0, len(d.Items))
-		for _, id := range d.Items {
-			intIds = append(intIds, int(id))
-		}
-		grain.Deliveries = append(grain.Deliveries, &CartDelivery{
-			Id:          int(d.Id),
-			Provider:    d.Provider,
-			Price:       d.Price,
-			Items:       intIds,
-			PickupPoint: d.PickupPoint,
-		})
-	}
-
-	return grain, nil
-}
-
-// GetCurrentState retrieves the current cart state using the typed StateReply oneof.
-func (g *RemoteGrainGRPC) GetCurrentState() (*CartGrain, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), g.stateTimeout)
-	defer cancel()
-	resp, err := g.client.GetState(ctx, &proto.StateRequest{CartId: g.Id.String()})
-	if err != nil {
-		return nil, err
-	}
-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		if e := resp.GetError(); e != "" {
-			return nil, fmt.Errorf("remote get state failed %d: %s", resp.StatusCode, e)
-		}
-		return nil, fmt.Errorf("remote get state failed %d", resp.StatusCode)
-	}
-	state := resp.GetState()
-	if state == nil {
-		return nil, fmt.Errorf("state reply missing state on success")
-	}
-	grain := &CartGrain{
-		Id:                ToCartId(state.Id),
-		TotalPrice:        state.TotalPrice,
-		TotalTax:          state.TotalTax,
-		TotalDiscount:     state.TotalDiscount,
-		PaymentInProgress: state.PaymentInProgress,
-		OrderReference:    state.OrderReference,
-		PaymentStatus:     state.PaymentStatus,
-	}
-	for _, it := range state.Items {
-		if it == nil {
-			continue
-		}
-		outlet := toPtr(it.Outlet)
-		storeId := toPtr(it.StoreId)
-		grain.Items = append(grain.Items, &CartItem{
-			Id:          int(it.Id),
-			ItemId:      int(it.ItemId),
-			Sku:         it.Sku,
-			Name:        it.Name,
-			Price:       it.Price,
-			Quantity:    int(it.Qty),
-			TotalPrice:  it.TotalPrice,
-			TotalTax:    it.TotalTax,
-			OrgPrice:    it.OrgPrice,
-			TaxRate:     int(it.TaxRate),
-			Brand:       it.Brand,
-			Category:    it.Category,
-			Category2:   it.Category2,
-			Category3:   it.Category3,
-			Category4:   it.Category4,
-			Category5:   it.Category5,
-			Image:       it.Image,
-			ArticleType: it.Type,
-			SellerId:    it.SellerId,
-			SellerName:  it.SellerName,
-			Disclaimer:  it.Disclaimer,
-			Outlet:      outlet,
-			StoreId:     storeId,
-			Stock:       StockStatus(it.Stock),
-		})
-	}
-	for _, d := range state.Deliveries {
-		if d == nil {
-			continue
-		}
-		intIds := make([]int, 0, len(d.Items))
-		for _, id := range d.Items {
-			intIds = append(intIds, int(id))
-		}
-		grain.Deliveries = append(grain.Deliveries, &CartDelivery{
-			Id:          int(d.Id),
-			Provider:    d.Provider,
-			Price:       d.Price,
-			Items:       intIds,
-			PickupPoint: d.PickupPoint,
-		})
-	}
-
-	return grain, nil
-}
-
-// Close closes the underlying gRPC connection if this adapter created it.
-func (g *RemoteGrainGRPC) Close() error {
-	if g.conn != nil {
-		return g.conn.Close()
-	}
-	return nil
-}
-
-// Debug helper to log operations (optional).
-func (g *RemoteGrainGRPC) logf(format string, args ...interface{}) {
-	log.Printf("[remote-grain-grpc host=%s id=%s] %s", g.Host, g.Id.String(), fmt.Sprintf(format, args...))
-}

ring.go

@@ -1,344 +0,0 @@
-package main
-
-import (
-	"encoding/binary"
-	"fmt"
-	"hash/fnv"
-	"sort"
-	"strings"
-	"sync"
-)
-
-// ring.go
-//
-// Consistent hashing ring skeleton for future integration.
-// --------------------------------------------------------
-// This file introduces a minimal, allocation‑light consistent hashing structure
-// intended to replace per-cart ownership negotiation. It focuses on:
-//   * Deterministic lookup: O(log V) via binary search
-//   * Even(ish) distribution using virtual nodes (vnodes)
-//   * Epoch / fingerprint tracking to detect membership drift
-//
-// NOT YET WIRED:
-//   * SyncedPool integration (ownerForCart, lazy migration)
-//   * Replication factor > 1
-//   * Persistent state migration
-//
-// Safe to import now; unused until explicit integration code is added.
-//
-// Design Notes
-// ------------
-// - Hosts contribute `vnodesPerHost` virtual nodes. Higher counts smooth
-//   distribution at cost of memory (V = hosts * vnodesPerHost).
-// - Hash of vnode = FNV1a64(host + "#" + index). For improved quality you
-//   can swap in xxhash or siphash later without changing API (but doing so
-//   will reshuffle ownership).
-// - Cart ownership lookup uses either cartID.Raw() when provided (uniform
-//   64-bit space) or falls back to hashing string forms (legacy).
-// - Epoch is monotonically increasing; consumers can fence stale results.
-//
-// Future Extensions
-// -----------------
-// - Weighted hosts (proportionally more vnodes).
-// - Replication: LookupN(h, n) to return primary + replicas.
-// - Streaming / diff-based ring updates (gossip).
-// - Hash function injection for deterministic test scenarios.
-//
-// ---------------------------------------------------------------------------
-
-// Vnode represents a single virtual node position on the ring.
-type Vnode struct {
-	Hash  uint64 // position on the ring
-	Host  string // physical host owning this vnode
-	Index int    // per-host vnode index (0..vnodesPerHost-1)
-}
-
-// Ring is an immutable consistent hash ring snapshot.
-type Ring struct {
-	Epoch       uint64
-	Vnodes      []Vnode // sorted by Hash
-	hosts       []string
-	fingerprint uint64 // membership fingerprint (order-independent)
-}
-
-// RingBuilder accumulates parameters to construct a Ring.
-type RingBuilder struct {
-	epoch         uint64
-	vnodesPerHost int
-	hosts         []string
-}
-
-// NewRingBuilder creates a builder with defaults.
-func NewRingBuilder() *RingBuilder {
-	return &RingBuilder{
-		vnodesPerHost: 64, // a reasonable default for small clusters
-	}
-}
-
-func (b *RingBuilder) WithEpoch(e uint64) *RingBuilder {
-	b.epoch = e
-	return b
-}
-
-func (b *RingBuilder) WithVnodesPerHost(n int) *RingBuilder {
-	if n > 0 {
-		b.vnodesPerHost = n
-	}
-	return b
-}
-
-func (b *RingBuilder) WithHosts(hosts []string) *RingBuilder {
-	uniq := make(map[string]struct{}, len(hosts))
-	out := make([]string, 0, len(hosts))
-	for _, h := range hosts {
-		h = strings.TrimSpace(h)
-		if h == "" {
-			continue
-		}
-		if _, ok := uniq[h]; ok {
-			continue
-		}
-		uniq[h] = struct{}{}
-		out = append(out, h)
-	}
-	sort.Strings(out)
-	b.hosts = out
-	return b
-}
-
-func (b *RingBuilder) Build() *Ring {
-	if len(b.hosts) == 0 {
-		return &Ring{
-			Epoch:       b.epoch,
-			Vnodes:      nil,
-			hosts:       nil,
-			fingerprint: 0,
-		}
-	}
-
-	totalVnodes := len(b.hosts) * b.vnodesPerHost
-	vnodes := make([]Vnode, 0, totalVnodes)
-
-	for _, host := range b.hosts {
-		for i := 0; i < b.vnodesPerHost; i++ {
-			h := hashVnode(host, i)
-			vnodes = append(vnodes, Vnode{
-				Hash:  h,
-				Host:  host,
-				Index: i,
-			})
-		}
-	}
-	sort.Slice(vnodes, func(i, j int) bool {
-		if vnodes[i].Hash == vnodes[j].Hash {
-			// Tie-break deterministically by host then index to avoid instability
-			if vnodes[i].Host == vnodes[j].Host {
-				return vnodes[i].Index < vnodes[j].Index
-			}
-			return vnodes[i].Host < vnodes[j].Host
-		}
-		return vnodes[i].Hash < vnodes[j].Hash
-	})
-
-	fp := fingerprintHosts(b.hosts)
-
-	return &Ring{
-		Epoch:       b.epoch,
-		Vnodes:      vnodes,
-		hosts:       append([]string(nil), b.hosts...),
-		fingerprint: fp,
-	}
-}
-
-// Hosts returns a copy of the host list (sorted).
-func (r *Ring) Hosts() []string {
-	if len(r.hosts) == 0 {
-		return nil
-	}
-	cp := make([]string, len(r.hosts))
-	copy(cp, r.hosts)
-	return cp
-}
-
-// Fingerprint returns a hash representing the unordered membership set.
-func (r *Ring) Fingerprint() uint64 {
-	return r.fingerprint
-}
-
-// Empty indicates ring has no vnodes.
-func (r *Ring) Empty() bool {
-	return len(r.Vnodes) == 0
-}
-
-// Lookup returns the vnode owning a given hash value.
-func (r *Ring) Lookup(h uint64) Vnode {
-	if len(r.Vnodes) == 0 {
-		return Vnode{}
-	}
-	// Binary search: first position with Hash >= h
-	i := sort.Search(len(r.Vnodes), func(i int) bool {
-		return r.Vnodes[i].Hash >= h
-	})
-	if i == len(r.Vnodes) {
-		return r.Vnodes[0]
-	}
-	return r.Vnodes[i]
-}
-
-// LookupID selects owner vnode for a CartID (fast path).
-func (r *Ring) LookupID(id CartID) Vnode {
-	return r.Lookup(id.Raw())
-}
-
-// LookupString hashes an arbitrary string and looks up owner.
-func (r *Ring) LookupString(s string) Vnode {
-	return r.Lookup(hashKeyString(s))
-}
-
-// LookupN returns up to n distinct host vnodes in ring order
-// starting from the primary owner of hash h (for replication).
-func (r *Ring) LookupN(h uint64, n int) []Vnode {
-	if n <= 0 || len(r.Vnodes) == 0 {
-		return nil
-	}
-	if n > len(r.hosts) {
-		n = len(r.hosts)
-	}
-	owners := make([]Vnode, 0, n)
-	seen := make(map[string]struct{}, n)
-
-	start := r.Lookup(h)
-
-	// Find index of start (can binary search again or linear scan; since we
-	// already have start.Hash we do another search for clarity)
-	i := sort.Search(len(r.Vnodes), func(i int) bool {
-		return r.Vnodes[i].Hash >= start.Hash
-	})
-	if i == len(r.Vnodes) {
-		i = 0
-	}
-
-	for idx := 0; len(owners) < n && idx < len(r.Vnodes); idx++ {
-		v := r.Vnodes[(i+idx)%len(r.Vnodes)]
-		if _, ok := seen[v.Host]; ok {
-			continue
-		}
-		seen[v.Host] = struct{}{}
-		owners = append(owners, v)
-	}
-	return owners
-}
-
-// DiffHosts compares this ring's membership to another.
-func (r *Ring) DiffHosts(other *Ring) (added []string, removed []string) {
-	if other == nil {
-		return r.Hosts(), nil
-	}
-	cur := make(map[string]struct{}, len(r.hosts))
-	for _, h := range r.hosts {
-		cur[h] = struct{}{}
-	}
-	oth := make(map[string]struct{}, len(other.hosts))
-	for _, h := range other.hosts {
-		oth[h] = struct{}{}
-	}
-	for h := range cur {
-		if _, ok := oth[h]; !ok {
-			removed = append(removed, h)
-		}
-	}
-	for h := range oth {
-		if _, ok := cur[h]; !ok {
-			added = append(added, h)
-		}
-	}
-	sort.Strings(added)
-	sort.Strings(removed)
-	return
-}
-
-// ---------------------------- Hash Functions ---------------------------------
-
-func hashVnode(host string, idx int) uint64 {
-	h := fnv.New64a()
-	_, _ = h.Write([]byte(host))
-	_, _ = h.Write([]byte{'#'})
-	var buf [8]byte
-	binary.BigEndian.PutUint64(buf[:], uint64(idx))
-	_, _ = h.Write(buf[:])
-	return h.Sum64()
-}
-
-// hashKeyString provides a stable hash for arbitrary string keys (legacy IDs).
-func hashKeyString(s string) uint64 {
-	h := fnv.New64a()
-	_, _ = h.Write([]byte(s))
-	return h.Sum64()
-}
-
-// fingerprintHosts produces an order-insensitive hash over the host set.
-func fingerprintHosts(hosts []string) uint64 {
-	if len(hosts) == 0 {
-		return 0
-	}
-	h := fnv.New64a()
-	for _, host := range hosts {
-		_, _ = h.Write([]byte(host))
-		_, _ = h.Write([]byte{0})
-	}
-	return h.Sum64()
-}
-
-// --------------------------- Thread-Safe Wrapper -----------------------------
-//
-// RingRef offers atomic swap + read semantics. SyncedPool can embed or hold
-// one of these to manage live ring updates safely.
-
-type RingRef struct {
-	mu   sync.RWMutex
-	ring *Ring
-}
-
-func NewRingRef(r *Ring) *RingRef {
-	return &RingRef{ring: r}
-}
-
-func (rr *RingRef) Get() *Ring {
-	rr.mu.RLock()
-	r := rr.ring
-	rr.mu.RUnlock()
-	return r
-}
-
-func (rr *RingRef) Set(r *Ring) {
-	rr.mu.Lock()
-	rr.ring = r
-	rr.mu.Unlock()
-}
-
-func (rr *RingRef) LookupID(id CartID) Vnode {
-	r := rr.Get()
-	if r == nil {
-		return Vnode{}
-	}
-	return r.LookupID(id)
-}
-
-// ----------------------------- Debug Utilities -------------------------------
-
-func (r *Ring) String() string {
-	var b strings.Builder
-	fmt.Fprintf(&b, "Ring{epoch=%d vnodes=%d hosts=%d}\n", r.Epoch, len(r.Vnodes), len(r.hosts))
-	limit := len(r.Vnodes)
-	if limit > 16 {
-		limit = 16
-	}
-	for i := 0; i < limit; i++ {
-		v := r.Vnodes[i]
-		fmt.Fprintf(&b, "  %02d hash=%016x host=%s idx=%d\n", i, v.Hash, v.Host, v.Index)
-	}
-	if len(r.Vnodes) > limit {
-		fmt.Fprintf(&b, "  ... (%d more)\n", len(r.Vnodes)-limit)
-	}
-	return b.String()
-}

synced-pool.go

@@ -1,647 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"reflect"
-	"sync"
-	"time"
-
-	proto "git.tornberg.me/go-cart-actor/proto"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-	"google.golang.org/grpc"
-	"k8s.io/apimachinery/pkg/watch"
-)
-
-// SyncedPool coordinates cart grain ownership across nodes using gRPC control plane
-// and cart actor services.
-//
-// Responsibilities:
-//   - Local grain access (delegates to GrainLocalPool)
-//   - Remote grain proxy management (RemoteGrainGRPC)
-//   - Cluster membership (AddRemote via discovery + negotiation)
-//   - Health/ping monitoring & remote removal
-//   - Ring based deterministic ownership (no runtime negotiation)
-//   - (Scaffolding) replication factor awareness via ring.LookupN
-//
-// Thread-safety: public methods that mutate internal maps lock p.mu (RWMutex).
-type SyncedPool struct {
-	Hostname string
-	local    *GrainLocalPool
-
-	mu sync.RWMutex
-
-	// Remote host state (gRPC only)
-	remoteHosts map[string]*RemoteHostGRPC // host -> remote host
-
-	// Remote grain proxies (by cart id)
-	remoteIndex map[CartId]Grain
-
-	// Discovery handler for re-adding hosts after failures
-	discardedHostHandler *DiscardedHostHandler
-
-	// Consistent hashing ring (immutable snapshot reference)
-	ringRef *RingRef
-
-	// Configuration
-	vnodesPerHost     int
-	replicationFactor int // RF (>=1). Currently only primary is active; replicas are scaffolding.
-}
-
-// RemoteHostGRPC tracks a remote host's clients & health.
-type RemoteHostGRPC struct {
-	Host          string
-	Conn          *grpc.ClientConn
-	CartClient    proto.CartActorClient
-	ControlClient proto.ControlPlaneClient
-	MissedPings   int
-}
-
-func (r *RemoteHostGRPC) IsHealthy() bool {
-	return r.MissedPings < 3
-}
-
-var (
-	negotiationCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "cart_remote_negotiation_total",
-		Help: "The total number of remote negotiations",
-	})
-	grainSyncCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "cart_grain_sync_total",
-		Help: "The total number of grain owner changes",
-	})
-	connectedRemotes = promauto.NewGauge(prometheus.GaugeOpts{
-		Name: "cart_connected_remotes",
-		Help: "The number of connected remotes",
-	})
-	remoteLookupCount = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "cart_remote_lookup_total",
-		Help: "The total number of remote lookups (legacy counter)",
-	})
-
-	// Ring / ownership metrics
-	ringEpoch = promauto.NewGauge(prometheus.GaugeOpts{
-		Name: "cart_ring_epoch",
-		Help: "Current consistent hashing ring epoch (fingerprint-based pseudo-epoch)",
-	})
-	ringHosts = promauto.NewGauge(prometheus.GaugeOpts{
-		Name: "cart_ring_hosts",
-		Help: "Number of hosts currently in the ring",
-	})
-	ringVnodes = promauto.NewGauge(prometheus.GaugeOpts{
-		Name: "cart_ring_vnodes",
-		Help: "Number of virtual nodes in the ring",
-	})
-	ringLookupLocal = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "cart_ring_lookup_local_total",
-		Help: "Ring ownership lookups resolved to the local host",
-	})
-	ringLookupRemote = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "cart_ring_lookup_remote_total",
-		Help: "Ring ownership lookups resolved to a remote host",
-	})
-	ringHostShare = promauto.NewGaugeVec(prometheus.GaugeOpts{
-		Name: "cart_ring_host_share",
-		Help: "Fractional share of ring vnodes per host",
-	}, []string{"host"})
-
-	cartMutationsTotal = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "cart_mutations_total",
-		Help: "Total number of cart state mutations applied (local + remote routed).",
-	})
-
-	cartMutationFailuresTotal = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "cart_mutation_failures_total",
-		Help: "Total number of failed cart state mutations (local apply errors or remote routing failures).",
-	})
-
-	cartMutationLatencySeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{
-		Name:    "cart_mutation_latency_seconds",
-		Help:    "Latency of cart mutations (successful or failed) in seconds.",
-		Buckets: prometheus.DefBuckets,
-	}, []string{"mutation"})
-
-	cartActiveGrains = promauto.NewGauge(prometheus.GaugeOpts{
-		Name: "cart_active_grains",
-		Help: "Number of active (resident) local grains.",
-	})
-)
-
-func NewSyncedPool(local *GrainLocalPool, hostname string, discovery Discovery) (*SyncedPool, error) {
-	p := &SyncedPool{
-		Hostname:             hostname,
-		local:                local,
-		remoteHosts:          make(map[string]*RemoteHostGRPC),
-		remoteIndex:          make(map[CartId]Grain),
-		discardedHostHandler: NewDiscardedHostHandler(1338),
-		vnodesPerHost:        64, // default smoothing factor; adjust if needed
-		replicationFactor:    1,  // RF scaffold; >1 not yet activating replicas
-	}
-	p.discardedHostHandler.SetReconnectHandler(p.AddRemote)
-	// Initialize empty ring (will be rebuilt after first AddRemote or discovery event)
-	p.rebuildRing()
-
-	if discovery != nil {
-		go func() {
-			time.Sleep(3 * time.Second) // allow gRPC server startup
-			log.Printf("Starting discovery watcher")
-			ch, err := discovery.Watch()
-			if err != nil {
-				log.Printf("Discovery error: %v", err)
-				return
-			}
-			for evt := range ch {
-				if evt.Host == "" {
-					continue
-				}
-				switch evt.Type {
-				case watch.Deleted:
-					if p.IsKnown(evt.Host) {
-						p.RemoveHost(evt.Host)
-					}
-				default:
-					if !p.IsKnown(evt.Host) {
-						log.Printf("Discovered host %s", evt.Host)
-						p.AddRemote(evt.Host)
-					}
-				}
-			}
-		}()
-	} else {
-		log.Printf("No discovery configured; expecting manual AddRemote or static host injection")
-	}
-
-	return p, nil
-}
-
-// ------------------------- Remote Host Management -----------------------------
-
-// AddRemote dials a remote host and initializes grain proxies.
-func (p *SyncedPool) AddRemote(host string) {
-	if host == "" || host == p.Hostname {
-		return
-	}
-
-	p.mu.Lock()
-	if _, exists := p.remoteHosts[host]; exists {
-		p.mu.Unlock()
-		return
-	}
-	p.mu.Unlock()
-
-	target := fmt.Sprintf("%s:1337", host)
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	conn, err := grpc.DialContext(ctx, target, grpc.WithInsecure(), grpc.WithBlock())
-	if err != nil {
-		log.Printf("AddRemote: dial %s failed: %v", target, err)
-		return
-	}
-
-	cartClient := proto.NewCartActorClient(conn)
-	controlClient := proto.NewControlPlaneClient(conn)
-
-	// Health check (Ping) with limited retries
-	pings := 3
-	for pings > 0 {
-		ctxPing, cancelPing := context.WithTimeout(context.Background(), 1*time.Second)
-		_, pingErr := controlClient.Ping(ctxPing, &proto.Empty{})
-		cancelPing()
-		if pingErr == nil {
-			break
-		}
-		pings--
-		time.Sleep(200 * time.Millisecond)
-		if pings == 0 {
-			log.Printf("AddRemote: ping %s failed after retries: %v", host, pingErr)
-			conn.Close()
-			return
-		}
-	}
-
-	remote := &RemoteHostGRPC{
-		Host:          host,
-		Conn:          conn,
-		CartClient:    cartClient,
-		ControlClient: controlClient,
-		MissedPings:   0,
-	}
-
-	p.mu.Lock()
-	p.remoteHosts[host] = remote
-	p.mu.Unlock()
-	connectedRemotes.Set(float64(p.RemoteCount()))
-	// Rebuild consistent hashing ring including this new host
-	p.rebuildRing()
-
-	log.Printf("Connected to remote host %s", host)
-
-	go p.pingLoop(remote)
-	go p.initializeRemote(remote)
-	go p.Negotiate()
-}
-
-// initializeRemote fetches remote cart ids and sets up remote grain proxies.
-func (p *SyncedPool) initializeRemote(remote *RemoteHostGRPC) {
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-	defer cancel()
-	reply, err := remote.ControlClient.GetCartIds(ctx, &proto.Empty{})
-	if err != nil {
-		log.Printf("Init remote %s: GetCartIds error: %v", remote.Host, err)
-		return
-	}
-	count := 0
-	for _, idStr := range reply.CartIds {
-		if idStr == "" {
-			continue
-		}
-		p.SpawnRemoteGrain(ToCartId(idStr), remote.Host)
-		count++
-	}
-	log.Printf("Remote %s reported %d grains", remote.Host, count)
-}
-
-// RemoveHost removes remote host and its grains.
-func (p *SyncedPool) RemoveHost(host string) {
-	p.mu.Lock()
-	remote, exists := p.remoteHosts[host]
-	if exists {
-		delete(p.remoteHosts, host)
-	}
-	// remove grains pointing to host
-	for id, g := range p.remoteIndex {
-		if rg, ok := g.(*RemoteGrainGRPC); ok && rg.Host == host {
-			delete(p.remoteIndex, id)
-		}
-	}
-	p.mu.Unlock()
-
-	if exists {
-		remote.Conn.Close()
-	}
-	connectedRemotes.Set(float64(p.RemoteCount()))
-	// Rebuild ring after host removal
-	p.rebuildRing()
-}
-
-// RemoteCount returns number of tracked remote hosts.
-func (p *SyncedPool) RemoteCount() int {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	return len(p.remoteHosts)
-}
-
-func (p *SyncedPool) IsKnown(host string) bool {
-	if host == p.Hostname {
-		return true
-	}
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	_, ok := p.remoteHosts[host]
-	return ok
-}
-
-func (p *SyncedPool) ExcludeKnown(hosts []string) []string {
-	ret := make([]string, 0, len(hosts))
-	for _, h := range hosts {
-		if !p.IsKnown(h) {
-			ret = append(ret, h)
-		}
-	}
-	return ret
-}
-
-// ------------------------- Health / Ping -------------------------------------
-
-func (p *SyncedPool) pingLoop(remote *RemoteHostGRPC) {
-	ticker := time.NewTicker(3 * time.Second)
-	defer ticker.Stop()
-	for range ticker.C {
-		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
-		_, err := remote.ControlClient.Ping(ctx, &proto.Empty{})
-		cancel()
-		if err != nil {
-			remote.MissedPings++
-			log.Printf("Ping %s failed (%d)", remote.Host, remote.MissedPings)
-			if !remote.IsHealthy() {
-				log.Printf("Remote %s unhealthy, removing", remote.Host)
-				p.RemoveHost(remote.Host)
-				return
-			}
-			continue
-		}
-		remote.MissedPings = 0
-	}
-}
-
-func (p *SyncedPool) IsHealthy() bool {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	for _, r := range p.remoteHosts {
-		if !r.IsHealthy() {
-			return false
-		}
-	}
-	return true
-}
-
-// ------------------------- Negotiation ---------------------------------------
-
-func (p *SyncedPool) Negotiate() {
-	negotiationCount.Inc()
-
-	p.mu.RLock()
-	hosts := make([]string, 0, len(p.remoteHosts)+1)
-	hosts = append(hosts, p.Hostname)
-	for h := range p.remoteHosts {
-		hosts = append(hosts, h)
-	}
-	remotes := make([]*RemoteHostGRPC, 0, len(p.remoteHosts))
-	for _, r := range p.remoteHosts {
-		remotes = append(remotes, r)
-	}
-	p.mu.RUnlock()
-
-	changed := false
-
-	for _, r := range remotes {
-		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-		reply, err := r.ControlClient.Negotiate(ctx, &proto.NegotiateRequest{KnownHosts: hosts})
-		cancel()
-		if err != nil {
-			log.Printf("Negotiate with %s failed: %v", r.Host, err)
-			continue
-		}
-		for _, h := range reply.Hosts {
-			if !p.IsKnown(h) {
-				p.AddRemote(h)
-				changed = true
-			}
-		}
-	}
-
-	// If new hosts were discovered during negotiation, rebuild the ring once at the end.
-	if changed {
-		p.rebuildRing()
-	}
-}
-
-// ------------------------- Grain / Ring Ownership ----------------------------
-
-// RemoveRemoteGrain removes a remote grain mapping.
-func (p *SyncedPool) RemoveRemoteGrain(id CartId) {
-	p.mu.Lock()
-	delete(p.remoteIndex, id)
-	p.mu.Unlock()
-}
-
-// SpawnRemoteGrain creates/updates a remote grain proxy for a given host.
-func (p *SyncedPool) SpawnRemoteGrain(id CartId, host string) {
-	if id.String() == "" {
-		return
-	}
-	p.mu.Lock()
-	// If local grain exists (legacy key), remove from local map (ownership moved).
-	if g, ok := p.local.grains[LegacyToCartKey(id)]; ok && g != nil {
-		delete(p.local.grains, LegacyToCartKey(id))
-	}
-	remoteHost, ok := p.remoteHosts[host]
-	if !ok {
-		p.mu.Unlock()
-		log.Printf("SpawnRemoteGrain: host %s unknown (id=%s), attempting AddRemote", host, id)
-		go p.AddRemote(host)
-		return
-	}
-	rg := NewRemoteGrainGRPC(id, host, remoteHost.CartClient)
-	p.remoteIndex[id] = rg
-	p.mu.Unlock()
-}
-
-// GetHealthyRemotes returns a copy slice of healthy remote hosts.
-func (p *SyncedPool) GetHealthyRemotes() []*RemoteHostGRPC {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	ret := make([]*RemoteHostGRPC, 0, len(p.remoteHosts))
-	for _, r := range p.remoteHosts {
-		if r.IsHealthy() {
-			ret = append(ret, r)
-		}
-	}
-	return ret
-}
-
-// rebuildRing reconstructs the consistent hashing ring from current host set
-// and updates ring-related metrics.
-func (p *SyncedPool) rebuildRing() {
-	p.mu.RLock()
-	hosts := make([]string, 0, len(p.remoteHosts)+1)
-	hosts = append(hosts, p.Hostname)
-	for h := range p.remoteHosts {
-		hosts = append(hosts, h)
-	}
-	p.mu.RUnlock()
-
-	epochSeed := fingerprintHosts(hosts)
-	builder := NewRingBuilder().
-		WithHosts(hosts).
-		WithEpoch(epochSeed).
-		WithVnodesPerHost(p.vnodesPerHost)
-	r := builder.Build()
-	if p.ringRef == nil {
-		p.ringRef = NewRingRef(r)
-	} else {
-		p.ringRef.Set(r)
-	}
-
-	// Metrics
-	ringEpoch.Set(float64(r.Epoch))
-	ringHosts.Set(float64(len(r.Hosts())))
-	ringVnodes.Set(float64(len(r.Vnodes)))
-	ringHostShare.Reset()
-	if len(r.Vnodes) > 0 {
-		perHost := make(map[string]int)
-		for _, v := range r.Vnodes {
-			perHost[v.Host]++
-		}
-		total := float64(len(r.Vnodes))
-		for h, c := range perHost {
-			ringHostShare.WithLabelValues(h).Set(float64(c) / total)
-		}
-	}
-}
-
-// ForceRingRefresh exposes a manual ring rebuild hook (primarily for tests).
-func (p *SyncedPool) ForceRingRefresh() {
-	p.rebuildRing()
-}
-
-// ownersFor returns the ordered list of primary + replica owners for a cart id
-// (length min(replicationFactor, #hosts)). Currently only the first (primary)
-// is used. This scaffolds future replication work.
-func (p *SyncedPool) ownersFor(id CartId) []string {
-	if p.ringRef == nil || p.replicationFactor <= 0 {
-		return []string{p.Hostname}
-	}
-	r := p.ringRef.Get()
-	if r == nil || r.Empty() {
-		return []string{p.Hostname}
-	}
-	vnodes := r.LookupN(hashKeyString(id.String()), p.replicationFactor)
-	out := make([]string, 0, len(vnodes))
-	seen := make(map[string]struct{}, len(vnodes))
-	for _, v := range vnodes {
-		if _, ok := seen[v.Host]; ok {
-			continue
-		}
-		seen[v.Host] = struct{}{}
-		out = append(out, v.Host)
-	}
-	if len(out) == 0 {
-		out = append(out, p.Hostname)
-	}
-	return out
-}
-
-// ownerHostFor returns the primary owner host for a given id.
-func (p *SyncedPool) ownerHostFor(id CartId) string {
-	return p.ownersFor(id)[0]
-}
-
-// DebugOwnerHost exposes (for tests) the currently computed primary owner host.
-func (p *SyncedPool) DebugOwnerHost(id CartId) string {
-	return p.ownerHostFor(id)
-}
-
-func (p *SyncedPool) removeLocalGrain(id CartId) {
-	p.mu.Lock()
-	delete(p.local.grains, LegacyToCartKey(id))
-	p.mu.Unlock()
-}
-
-// getGrain returns a local or remote grain. For remote ownership it performs a
-// bounded readiness wait (small retries) to reduce first-call failures while
-// the remote connection & proxy are initializing.
-func (p *SyncedPool) getGrain(id CartId) (Grain, error) {
-	owner := p.ownerHostFor(id)
-	if owner == p.Hostname {
-		ringLookupLocal.Inc()
-		grain, err := p.local.GetGrain(id)
-		if err != nil {
-			return nil, err
-		}
-		return grain, nil
-	}
-	ringLookupRemote.Inc()
-
-	// Kick off remote dial if we don't yet know the owner.
-	if !p.IsKnown(owner) {
-		go p.AddRemote(owner)
-	}
-
-	// Fast path existing proxy
-	p.mu.RLock()
-	if rg, ok := p.remoteIndex[id]; ok {
-		p.mu.RUnlock()
-		remoteLookupCount.Inc()
-		return rg, nil
-	}
-	p.mu.RUnlock()
-
-	const (
-		attempts    = 5
-		sleepPerTry = 40 * time.Millisecond
-	)
-
-	for attempt := 0; attempt < attempts; attempt++ {
-		// Try to spawn (idempotent if host already known)
-		if p.IsKnown(owner) {
-			p.SpawnRemoteGrain(id, owner)
-		}
-		// Check again
-		p.mu.RLock()
-		if rg, ok := p.remoteIndex[id]; ok {
-			p.mu.RUnlock()
-			remoteLookupCount.Inc()
-			return rg, nil
-		}
-		p.mu.RUnlock()
-
-		// Last attempt? break to return error.
-		if attempt == attempts-1 {
-			break
-		}
-		time.Sleep(sleepPerTry)
-	}
-
-	return nil, fmt.Errorf("remote owner %s not yet available for cart %s (after %d attempts)", owner, id.String(), attempts)
-}
-
-// Apply applies a single mutation to a grain (local or remote).
-// Replication (RF>1) scaffolding: future enhancement will fan-out mutations
-// to replica owners (best-effort) and reconcile quorum on read.
-func (p *SyncedPool) Apply(id CartId, mutation interface{}) (*CartGrain, error) {
-	grain, err := p.getGrain(id)
-	if err != nil {
-		return nil, err
-	}
-	start := time.Now()
-	result, applyErr := grain.Apply(mutation, false)
-
-	// Derive mutation type label (strip pointer)
-	mutationType := "unknown"
-	if mutation != nil {
-		if t := reflect.TypeOf(mutation); t != nil {
-			if t.Kind() == reflect.Ptr {
-				t = t.Elem()
-			}
-			if t.Name() != "" {
-				mutationType = t.Name()
-			}
-		}
-	}
-	cartMutationLatencySeconds.WithLabelValues(mutationType).Observe(time.Since(start).Seconds())
-
-	if applyErr == nil && result != nil {
-		cartMutationsTotal.Inc()
-		if p.ownerHostFor(id) == p.Hostname {
-			// Update active grains gauge only for local ownership
-			cartActiveGrains.Set(float64(p.local.DebugGrainCount()))
-		}
-	} else if applyErr != nil {
-		cartMutationFailuresTotal.Inc()
-	}
-	return result, applyErr
-}
-
-// Get returns current state of a grain (local or remote).
-// Future replication hook: Read-repair or quorum read can be added here.
-func (p *SyncedPool) Get(id CartId) (*CartGrain, error) {
-	grain, err := p.getGrain(id)
-	if err != nil {
-		return nil, err
-	}
-	return grain.GetCurrentState()
-}
-
-// Close notifies remotes this host is terminating.
-func (p *SyncedPool) Close() {
-	p.mu.RLock()
-	remotes := make([]*RemoteHostGRPC, 0, len(p.remoteHosts))
-	for _, r := range p.remoteHosts {
-		remotes = append(remotes, r)
-	}
-	p.mu.RUnlock()
-
-	for _, r := range remotes {
-		go func(rh *RemoteHostGRPC) {
-			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
-			_, err := rh.ControlClient.Closing(ctx, &proto.ClosingNotice{Host: p.Hostname})
-			cancel()
-			if err != nil {
-				log.Printf("Close notify to %s failed: %v", rh.Host, err)
-			}
-		}(r)
-	}
-}

tcp-connection_test.go

@@ -1,8 +0,0 @@
-/*
-Legacy TCP networking (GenericListener / Frame protocol) has been removed
-as part of the gRPC migration. This file intentionally contains no tests.
-
-Keeping an empty Go file (with a package declaration) ensures the old
-tcp-connection test target no longer runs without causing build issues.
-*/
-package main