Complete refactor to new grpc control plane and only http proxy for carts (#4)

Co-authored-by: matst80 <mats.tornberg@gmail.com>
Reviewed-on: https://git.tornberg.me/mats/go-cart-actor/pulls/4
Co-authored-by: Mats Törnberg <mats@tornberg.me>
Co-committed-by: Mats Törnberg <mats@tornberg.me>

pkg/actor/disk_storage.go

@@ -0,0 +1,137 @@
+package actor
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+)
+
+type QueueEvent struct {
+	TimeStamp time.Time
+	Message   proto.Message
+}
+
+type DiskStorage[V any] struct {
+	*StateStorage
+	path  string
+	done  chan struct{}
+	queue *sync.Map // map[uint64][]QueueEvent
+}
+
+type LogStorage[V any] interface {
+	LoadEvents(id uint64, grain Grain[V]) error
+	AppendEvent(id uint64, msg ...proto.Message) error
+}
+
+func NewDiskStorage[V any](path string, registry MutationRegistry) *DiskStorage[V] {
+	return &DiskStorage[V]{
+		StateStorage: NewState(registry),
+		path:         path,
+		done:         make(chan struct{}),
+	}
+}
+
+func (s *DiskStorage[V]) SaveLoop(duration time.Duration) {
+	s.queue = &sync.Map{}
+	ticker := time.NewTicker(duration)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-s.done:
+			s.save()
+			return
+		case <-ticker.C:
+			s.save()
+		}
+	}
+}
+
+func (s *DiskStorage[V]) save() {
+	carts := 0
+	lines := 0
+	s.queue.Range(func(key, value any) bool {
+		id := key.(uint64)
+		path := s.logPath(id)
+		fh, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+		if err != nil {
+			log.Printf("failed to open event log file: %v", err)
+			return true
+		}
+		defer fh.Close()
+
+		if qe, ok := value.([]QueueEvent); ok {
+			for _, msg := range qe {
+				if err := s.Append(fh, msg.Message, msg.TimeStamp); err != nil {
+					log.Printf("failed to append event to log file: %v", err)
+				}
+				lines++
+			}
+		}
+		carts++
+		s.queue.Delete(id)
+		return true
+	})
+	if lines > 0 {
+		log.Printf("Appended %d carts and %d lines to disk", carts, lines)
+	}
+}
+
+func (s *DiskStorage[V]) logPath(id uint64) string {
+	return filepath.Join(s.path, fmt.Sprintf("%d.events.log", id))
+}
+
+func (s *DiskStorage[V]) LoadEvents(id uint64, grain Grain[V]) error {
+	path := s.logPath(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 {
+		return fmt.Errorf("open replay file: %w", err)
+	}
+	defer fh.Close()
+	return s.Load(fh, func(msg proto.Message) {
+		s.registry.Apply(grain, msg)
+	})
+}
+
+func (s *DiskStorage[V]) Close() {
+	s.save()
+	close(s.done)
+}
+
+func (s *DiskStorage[V]) AppendEvent(id uint64, msg ...proto.Message) error {
+	if s.queue != nil {
+		queue := make([]QueueEvent, 0)
+		data, found := s.queue.Load(id)
+		if found {
+			queue = data.([]QueueEvent)
+		}
+		for _, m := range msg {
+			queue = append(queue, QueueEvent{Message: m, TimeStamp: time.Now()})
+		}
+		s.queue.Store(id, queue)
+		return nil
+	} else {
+		path := s.logPath(id)
+		fh, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+		if err != nil {
+			log.Printf("failed to open event log file: %v", err)
+			return err
+		}
+		defer fh.Close()
+		for _, m := range msg {
+			err = s.Append(fh, m, time.Now())
+		}
+		return err
+	}
+
+}

pkg/actor/grain.go

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

pkg/actor/grain_pool.go

@@ -0,0 +1,41 @@
+package actor
+
+import (
+	"net/http"
+
+	"github.com/gogo/protobuf/proto"
+)
+
+type MutationResult[V any] struct {
+	Result    V             `json:"result"`
+	Mutations []ApplyResult `json:"mutations,omitempty"`
+}
+
+type GrainPool[V any] interface {
+	Apply(id uint64, mutation ...proto.Message) (*MutationResult[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
+	IsKnown(string) bool
+	Close()
+}
+
+// Host abstracts a remote node capable of proxying cart requests.
+type Host interface {
+	AnnounceExpiry(ids []uint64)
+	Negotiate(otherHosts []string) ([]string, error)
+	Name() string
+	Proxy(id uint64, w http.ResponseWriter, r *http.Request) (bool, error)
+	GetActorIds() []uint64
+	Close() error
+	Ping() bool
+	IsHealthy() bool
+	AnnounceOwnership(ownerHost string, ids []uint64)
+}

pkg/actor/grpc_server.go

@@ -0,0 +1,119 @@
+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) {
+	// log.Printf("got ping")
+	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
+}
+
+type ServerConfig struct {
+	Addr    string
+	Options []grpc.ServerOption
+}
+
+func NewServerConfig(addr string, options ...grpc.ServerOption) ServerConfig {
+	return ServerConfig{
+		Addr:    addr,
+		Options: options,
+	}
+}
+
+func DefaultServerConfig() ServerConfig {
+	return NewServerConfig(":1337")
+}
+
+// StartGRPCServer configures and starts the unified gRPC server on the given address.
+// It registers both the CartActor and ControlPlane services.
+func NewControlServer[V any](config ServerConfig, pool GrainPool[V]) (*grpc.Server, error) {
+	lis, err := net.Listen("tcp", config.Addr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to listen: %w", err)
+	}
+
+	grpcServer := grpc.NewServer(config.Options...)
+	server := &ControlServer[V]{
+		pool: pool,
+	}
+
+	messages.RegisterControlPlaneServer(grpcServer, server)
+	reflection.Register(grpcServer)
+
+	log.Printf("gRPC server listening as %s on %s", pool.Hostname(), config.Addr)
+	go func() {
+		if err := grpcServer.Serve(lis); err != nil {
+			log.Fatalf("failed to serve gRPC: %v", err)
+		}
+	}()
+
+	return grpcServer, nil
+}

pkg/actor/mutation_registry.go

@@ -0,0 +1,226 @@
+package actor
+
+import (
+	"fmt"
+	"log"
+	"reflect"
+	"sync"
+
+	"github.com/gogo/protobuf/proto"
+)
+
+type ApplyResult struct {
+	Type     string        `json:"type"`
+	Mutation proto.Message `json:"mutation"`
+	Error    error         `json:"error,omitempty"`
+}
+
+type MutationRegistry interface {
+	Apply(grain any, msg ...proto.Message) ([]ApplyResult, error)
+	RegisterMutations(handlers ...MutationHandler)
+	Create(typeName string) (proto.Message, bool)
+	GetTypeName(msg proto.Message) (string, bool)
+	//GetStorageEvent(msg proto.Message) StorageEvent
+	//FromStorageEvent(event StorageEvent) (proto.Message, error)
+}
+
+type ProtoMutationRegistry struct {
+	mutationRegistryMu sync.RWMutex
+	mutationRegistry   map[reflect.Type]MutationHandler
+}
+
+var (
+	ErrMutationNotRegistered = &MutationError{
+		Message:    "mutation not registered",
+		Code:       255,
+		StatusCode: 500,
+	}
+)
+
+type MutationError struct {
+	Message    string `json:"message"`
+	Code       uint32 `json:"code"`
+	StatusCode uint32 `json:"status_code"`
+}
+
+func (m MutationError) Error() string {
+	return m.Message
+}
+
+// MutationOption configures additional behavior for a registered mutation.
+type MutationOption func(*mutationOptions)
+
+// mutationOptions holds flags adjustable per registration.
+type mutationOptions struct {
+	updateTotals bool
+}
+
+// WithTotals ensures CartGrain.UpdateTotals() is called after a successful handler.
+func WithTotals() MutationOption {
+	return func(o *mutationOptions) {
+		o.updateTotals = true
+	}
+}
+
+type MutationHandler interface {
+	Handle(state any, msg proto.Message) error
+	Name() string
+	Type() reflect.Type
+	Create() proto.Message
+}
+
+// RegisteredMutation stores metadata + the execution closure.
+type RegisteredMutation[V any, T proto.Message] struct {
+	name    string
+	handler func(*V, T) error
+	create  func() T
+	msgType reflect.Type
+}
+
+func NewMutation[V any, T proto.Message](handler func(*V, T) error, create func() T) *RegisteredMutation[V, T] {
+	// Derive the name and message type from a concrete instance produced by create().
+	// This avoids relying on reflect.TypeFor (which can yield unexpected results in some toolchains)
+	// and ensures we always peel off the pointer layer for proto messages.
+	instance := create()
+	rt := reflect.TypeOf(instance)
+	if rt.Kind() == reflect.Ptr {
+		rt = rt.Elem()
+	}
+	return &RegisteredMutation[V, T]{
+		name:    rt.Name(),
+		handler: handler,
+		create:  create,
+		msgType: rt,
+	}
+}
+
+func (m *RegisteredMutation[V, T]) Handle(state any, msg proto.Message) error {
+	return m.handler(state.(*V), msg.(T))
+}
+
+func (m *RegisteredMutation[V, T]) Name() string {
+	return m.name
+}
+
+func (m *RegisteredMutation[V, T]) Create() proto.Message {
+	return m.create()
+}
+
+func (m *RegisteredMutation[V, T]) Type() reflect.Type {
+	return m.msgType
+}
+
+func NewMutationRegistry() MutationRegistry {
+	return &ProtoMutationRegistry{
+		mutationRegistry:   make(map[reflect.Type]MutationHandler),
+		mutationRegistryMu: sync.RWMutex{},
+	}
+}
+
+func (r *ProtoMutationRegistry) RegisterMutations(handlers ...MutationHandler) {
+	r.mutationRegistryMu.Lock()
+	defer r.mutationRegistryMu.Unlock()
+
+	for _, handler := range handlers {
+		r.mutationRegistry[handler.Type()] = handler
+	}
+}
+
+func (r *ProtoMutationRegistry) GetTypeName(msg proto.Message) (string, bool) {
+	r.mutationRegistryMu.RLock()
+	defer r.mutationRegistryMu.RUnlock()
+
+	rt := indirectType(reflect.TypeOf(msg))
+	if handler, ok := r.mutationRegistry[rt]; ok {
+		return handler.Name(), true
+	}
+	return "", false
+}
+
+func (r *ProtoMutationRegistry) getHandler(typeName string) MutationHandler {
+	r.mutationRegistryMu.Lock()
+	defer r.mutationRegistryMu.Unlock()
+
+	for _, handler := range r.mutationRegistry {
+		if handler.Name() == typeName {
+			return handler
+		}
+	}
+
+	return nil
+}
+
+func (r *ProtoMutationRegistry) Create(typeName string) (proto.Message, bool) {
+
+	handler := r.getHandler(typeName)
+	if handler == nil {
+		log.Printf("missing handler for %s", typeName)
+		return nil, false
+	}
+
+	return handler.Create(), true
+}
+
+// ApplyRegistered attempts to apply a registered mutation.
+// Returns updated grain if successful.
+//
+// If the mutation is not registered, returns (nil, ErrMutationNotRegistered).
+func (r *ProtoMutationRegistry) Apply(grain any, msg ...proto.Message) ([]ApplyResult, error) {
+	results := make([]ApplyResult, 0, len(msg))
+
+	if grain == nil {
+		return results, fmt.Errorf("nil grain")
+	}
+	if msg == nil {
+		return results, fmt.Errorf("nil mutation message")
+	}
+
+	for _, m := range msg {
+		rt := indirectType(reflect.TypeOf(m))
+		r.mutationRegistryMu.RLock()
+		entry, ok := r.mutationRegistry[rt]
+		r.mutationRegistryMu.RUnlock()
+		if !ok {
+			results = append(results, ApplyResult{Error: ErrMutationNotRegistered, Type: rt.Name(), Mutation: m})
+			continue
+		}
+		err := entry.Handle(grain, m)
+		results = append(results, ApplyResult{Error: err, Type: rt.Name(), Mutation: m})
+	}
+
+	// if entry.updateTotals {
+	// 	grain.UpdateTotals()
+	// }
+
+	return results, nil
+}
+
+// RegisteredMutations returns metadata for all registered mutations (snapshot).
+func (r *ProtoMutationRegistry) RegisteredMutations() []string {
+	r.mutationRegistryMu.RLock()
+	defer r.mutationRegistryMu.RUnlock()
+	out := make([]string, 0, len(r.mutationRegistry))
+	for _, entry := range r.mutationRegistry {
+		out = append(out, entry.Name())
+	}
+	return out
+}
+
+// RegisteredMutationTypes returns the reflect.Type list of all registered messages.
+// Useful for coverage tests ensuring expected set matches actual set.
+func (r *ProtoMutationRegistry) RegisteredMutationTypes() []reflect.Type {
+	r.mutationRegistryMu.RLock()
+	defer r.mutationRegistryMu.RUnlock()
+	out := make([]reflect.Type, 0, len(r.mutationRegistry))
+	for _, entry := range r.mutationRegistry {
+		out = append(out, entry.Type())
+	}
+	return out
+}
+
+func indirectType(t reflect.Type) reflect.Type {
+	for t.Kind() == reflect.Ptr {
+		t = t.Elem()
+	}
+	return t
+}

pkg/actor/mutation_registry_test.go

@@ -0,0 +1,134 @@
+package actor
+
+import (
+	"errors"
+	"reflect"
+	"slices"
+	"testing"
+
+	"git.tornberg.me/go-cart-actor/pkg/messages"
+)
+
+type cartState struct {
+	calls     int
+	lastAdded *messages.AddItem
+}
+
+func TestRegisteredMutationBasics(t *testing.T) {
+	reg := NewMutationRegistry().(*ProtoMutationRegistry)
+
+	addItemMutation := NewMutation(
+		func(state *cartState, msg *messages.AddItem) error {
+			state.calls++
+			// copy to avoid external mutation side-effects (not strictly necessary for the test)
+			cp := *msg
+			state.lastAdded = &cp
+			return nil
+		},
+		func() *messages.AddItem { return &messages.AddItem{} },
+	)
+
+	// Sanity check on mutation metadata
+	if addItemMutation.Name() != "AddItem" {
+		t.Fatalf("expected mutation Name() == AddItem, got %s", addItemMutation.Name())
+	}
+	if got, want := addItemMutation.Type(), reflect.TypeOf(messages.AddItem{}); got != want {
+		t.Fatalf("expected Type() == %v, got %v", want, got)
+	}
+
+	reg.RegisterMutations(addItemMutation)
+
+	// RegisteredMutations: membership (order not guaranteed)
+	names := reg.RegisteredMutations()
+	if !slices.Contains(names, "AddItem") {
+		t.Fatalf("RegisteredMutations missing AddItem, got %v", names)
+	}
+
+	// RegisteredMutationTypes: membership (order not guaranteed)
+	types := reg.RegisteredMutationTypes()
+	if !slices.Contains(types, reflect.TypeOf(messages.AddItem{})) {
+		t.Fatalf("RegisteredMutationTypes missing AddItem type, got %v", types)
+	}
+
+	// GetTypeName should resolve for a pointer instance
+	name, ok := reg.GetTypeName(&messages.AddItem{})
+	if !ok || name != "AddItem" {
+		t.Fatalf("GetTypeName returned (%q,%v), expected (AddItem,true)", name, ok)
+	}
+
+	// GetTypeName should fail for unregistered type
+	if name, ok := reg.GetTypeName(&messages.Noop{}); ok || name != "" {
+		t.Fatalf("expected GetTypeName to fail for unregistered message, got (%q,%v)", name, ok)
+	}
+
+	// Create by name
+	msg, ok := reg.Create("AddItem")
+	if !ok {
+		t.Fatalf("Create failed for registered mutation")
+	}
+	if _, isAddItem := msg.(*messages.AddItem); !isAddItem {
+		t.Fatalf("Create returned wrong concrete type: %T", msg)
+	}
+
+	// Create unknown
+	if m2, ok := reg.Create("Unknown"); ok || m2 != nil {
+		t.Fatalf("Create should fail for unknown mutation, got (%T,%v)", m2, ok)
+	}
+
+	// Apply happy path
+	state := &cartState{}
+	add := &messages.AddItem{ItemId: 42, Quantity: 3, Sku: "ABC"}
+	if _, err := reg.Apply(state, add); err != nil {
+		t.Fatalf("Apply returned error: %v", err)
+	}
+	if state.calls != 1 {
+		t.Fatalf("handler not invoked expected calls=1 got=%d", state.calls)
+	}
+	if state.lastAdded == nil || state.lastAdded.ItemId != 42 || state.lastAdded.Quantity != 3 {
+		t.Fatalf("state not updated correctly: %+v", state.lastAdded)
+	}
+
+	// Apply nil grain
+	if _, err := reg.Apply(nil, add); err == nil {
+		t.Fatalf("expected error for nil grain")
+	}
+
+	// Apply nil message
+	if _, err := reg.Apply(state, nil); err == nil {
+		t.Fatalf("expected error for nil mutation message")
+	}
+
+	// Apply unregistered message
+	if _, err := reg.Apply(state, &messages.Noop{}); !errors.Is(err, ErrMutationNotRegistered) {
+		t.Fatalf("expected ErrMutationNotRegistered, got %v", err)
+	}
+}
+
+// func TestConcurrentSafeRegistrationLookup(t *testing.T) {
+// 	// This test is light-weight; it ensures locks don't deadlock under simple concurrent access.
+// 	reg := NewMutationRegistry().(*ProtoMutationRegistry)
+// 	mut := NewMutation[cartState, *messages.Noop](
+// 		func(state *cartState, msg *messages.Noop) error { state.calls++; return nil },
+// 		func() *messages.Noop { return &messages.Noop{} },
+// 	)
+// 	reg.RegisterMutations(mut)
+
+// 	done := make(chan struct{})
+// 	const workers = 25
+// 	for i := 0; i < workers; i++ {
+// 		go func() {
+// 			for j := 0; j < 100; j++ {
+// 				_, _ = reg.Create("Noop")
+// 				_, _ = reg.GetTypeName(&messages.Noop{})
+// 				_ = reg.Apply(&cartState{}, &messages.Noop{})
+// 			}
+// 			done <- struct{}{}
+// 		}()
+// 	}
+
+// 	for i := 0; i < workers; i++ {
+// 		<-done
+// 	}
+// }
+
+// Helpers

pkg/actor/simple_grain_pool.go

@@ -0,0 +1,433 @@
+package actor
+
+import (
+	"fmt"
+	"log"
+	"maps"
+	"sync"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+)
+
+type SimpleGrainPool[V any] struct {
+	// fields and methods
+	localMu          sync.RWMutex
+	grains           map[uint64]Grain[V]
+	mutationRegistry MutationRegistry
+	spawn            func(id uint64) (Grain[V], error)
+	spawnHost        func(host string) (Host, error)
+	storage          LogStorage[V]
+	ttl              time.Duration
+	poolSize         int
+
+	// Cluster coordination --------------------------------------------------
+	hostname     string
+	remoteMu     sync.RWMutex
+	remoteOwners map[uint64]Host
+	remoteHosts  map[string]Host
+	//discardedHostHandler *DiscardedHostHandler
+
+	// House-keeping ---------------------------------------------------------
+	purgeTicker *time.Ticker
+}
+
+type GrainPoolConfig[V any] struct {
+	Hostname         string
+	Spawn            func(id uint64) (Grain[V], error)
+	SpawnHost        func(host string) (Host, error)
+	TTL              time.Duration
+	PoolSize         int
+	MutationRegistry MutationRegistry
+	Storage          LogStorage[V]
+}
+
+func NewSimpleGrainPool[V any](config GrainPoolConfig[V]) (*SimpleGrainPool[V], error) {
+	p := &SimpleGrainPool[V]{
+		grains:           make(map[uint64]Grain[V]),
+		mutationRegistry: config.MutationRegistry,
+		storage:          config.Storage,
+		spawn:            config.Spawn,
+		spawnHost:        config.SpawnHost,
+		ttl:              config.TTL,
+		poolSize:         config.PoolSize,
+		hostname:         config.Hostname,
+		remoteOwners:     make(map[uint64]Host),
+		remoteHosts:      make(map[string]Host),
+	}
+
+	p.purgeTicker = time.NewTicker(time.Minute)
+	go func() {
+		for range p.purgeTicker.C {
+			p.purge()
+		}
+	}()
+
+	return p, nil
+}
+
+func (p *SimpleGrainPool[V]) 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 Host) {
+		remote.AnnounceExpiry(purgedIds)
+	})
+
+}
+
+// LocalUsage returns the number of resident grains and configured capacity.
+func (p *SimpleGrainPool[V]) 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 *SimpleGrainPool[V]) 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 *SimpleGrainPool[V]) 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 *SimpleGrainPool[V]) HandleOwnershipChange(host string, ids []uint64) error {
+	log.Printf("host %s now owns %d cart ids", host, len(ids))
+	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
+}
+
+// TakeOwnership takes ownership of a grain.
+func (p *SimpleGrainPool[V]) TakeOwnership(id uint64) {
+	p.broadcastOwnership([]uint64{id})
+}
+
+func (p *SimpleGrainPool[V]) AddRemote(host string) (Host, error) {
+	if host == "" {
+		return nil, fmt.Errorf("host is empty")
+	}
+	if host == p.hostname {
+		return nil, fmt.Errorf("same host, this should not happen")
+	}
+	p.remoteMu.RLock()
+	existing, found := p.remoteHosts[host]
+	p.remoteMu.RUnlock()
+	if found {
+		return existing, nil
+	}
+
+	remote, err := p.spawnHost(host)
+	if err != nil {
+		log.Printf("AddRemote %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 *SimpleGrainPool[V]) initializeRemote(remote Host) {
+
+	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 *SimpleGrainPool[V]) RemoveHost(host string) {
+	p.remoteMu.Lock()
+	remote, exists := p.remoteHosts[host]
+
+	if exists {
+		go remote.Close()
+		delete(p.remoteHosts, host)
+	}
+	count := 0
+	for id, owner := range p.remoteOwners {
+		if owner.Name() == host {
+			count++
+			delete(p.remoteOwners, id)
+		}
+	}
+	log.Printf("Removing host %s, grains: %d", host, count)
+	p.remoteMu.Unlock()
+
+	if exists {
+		remote.Close()
+	}
+	// connectedRemotes.Set(float64(p.RemoteCount()))
+}
+
+func (p *SimpleGrainPool[V]) RemoteCount() int {
+	p.remoteMu.RLock()
+	defer p.remoteMu.RUnlock()
+	return len(p.remoteHosts)
+}
+
+// RemoteHostNames returns a snapshot of connected remote host identifiers.
+func (p *SimpleGrainPool[V]) 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 *SimpleGrainPool[V]) 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 *SimpleGrainPool[V]) pingLoop(remote Host) {
+	remote.Ping()
+	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.Name())
+				p.Close()
+				p.RemoveHost(remote.Name())
+				return
+			}
+			continue
+		}
+	}
+}
+
+func (p *SimpleGrainPool[V]) IsHealthy() bool {
+	p.remoteMu.RLock()
+	defer p.remoteMu.RUnlock()
+	for _, r := range p.remoteHosts {
+		if !r.IsHealthy() {
+			return false
+		}
+	}
+	return true
+}
+
+func (p *SimpleGrainPool[V]) 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 *SimpleGrainPool[V]) SendNegotiation() {
+	//negotiationCount.Inc()
+
+	p.remoteMu.RLock()
+	hosts := make([]string, 0, len(p.remoteHosts)+1)
+	hosts = append(hosts, p.hostname)
+	remotes := make([]Host, 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 Host) {
+		knownByRemote, err := remote.Negotiate(hosts)
+
+		if err != nil {
+			log.Printf("Negotiate with %s failed: %v", remote.Name(), err)
+			return
+		}
+		for _, h := range knownByRemote {
+			if !p.IsKnown(h) {
+				go p.AddRemote(h)
+			}
+		}
+	})
+}
+
+func (p *SimpleGrainPool[V]) forAllHosts(fn func(Host)) {
+	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 *SimpleGrainPool[V]) broadcastOwnership(ids []uint64) {
+	if len(ids) == 0 {
+		return
+	}
+
+	p.forAllHosts(func(rh Host) {
+		rh.AnnounceOwnership(p.hostname, ids)
+	})
+	log.Printf("%s taking ownership of %d ids", p.hostname, len(ids))
+	// go p.statsUpdate()
+}
+
+func (p *SimpleGrainPool[V]) getOrClaimGrain(id uint64) (Grain[V], error) {
+	p.localMu.RLock()
+	grain, exists := p.grains[id]
+	p.localMu.RUnlock()
+	if exists && grain != nil {
+		return grain, nil
+	}
+
+	grain, err := p.spawn(id)
+	if err != nil {
+		return nil, err
+	}
+	p.localMu.Lock()
+	p.grains[id] = grain
+	p.localMu.Unlock()
+	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 *SimpleGrainPool[V]) Apply(id uint64, mutation ...proto.Message) (*MutationResult[*V], error) {
+	grain, err := p.getOrClaimGrain(id)
+	if err != nil {
+		return nil, err
+	}
+
+	mutations, err := p.mutationRegistry.Apply(grain, mutation...)
+	if err != nil {
+		return nil, err
+	}
+	if p.storage != nil {
+		go func() {
+			if err := p.storage.AppendEvent(id, mutation...); err != nil {
+				log.Printf("failed to store mutation for grain %d: %v", id, err)
+			}
+		}()
+	}
+	result, err := grain.GetCurrentState()
+	if err != nil {
+		return nil, err
+	}
+	return &MutationResult[*V]{
+		Result:    result,
+		Mutations: mutations,
+	}, nil
+}
+
+// Get returns the current state of a grain.
+func (p *SimpleGrainPool[V]) Get(id uint64) (*V, 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 *SimpleGrainPool[V]) OwnerHost(id uint64) (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 *SimpleGrainPool[V]) Hostname() string {
+	return p.hostname
+}
+
+// Close notifies remotes that this host is shutting down.
+func (p *SimpleGrainPool[V]) Close() {
+
+	p.forAllHosts(func(rh Host) {
+		rh.Close()
+	})
+
+	if p.purgeTicker != nil {
+		p.purgeTicker.Stop()
+	}
+}

pkg/actor/state.go

@@ -0,0 +1,98 @@
+package actor
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"io"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+)
+
+type StateStorage struct {
+	registry MutationRegistry
+}
+
+type StorageEvent struct {
+	Type      string        `json:"type"`
+	TimeStamp time.Time     `json:"timestamp"`
+	Mutation  proto.Message `json:"mutation"`
+}
+
+type rawEvent struct {
+	Type      string          `json:"type"`
+	TimeStamp time.Time       `json:"timestamp"`
+	Mutation  json.RawMessage `json:"mutation"`
+}
+
+func NewState(registry MutationRegistry) *StateStorage {
+	return &StateStorage{
+		registry: registry,
+	}
+}
+
+var ErrUnknownType = errors.New("unknown type")
+
+func (s *StateStorage) Load(r io.Reader, onMessage func(msg proto.Message)) error {
+	var err error
+	var evt *StorageEvent
+	scanner := bufio.NewScanner(r)
+	for err == nil {
+		evt, err = s.Read(scanner)
+		if err == nil {
+			onMessage(evt.Mutation)
+		}
+	}
+	if err == io.EOF {
+		return nil
+	}
+	return err
+}
+
+func (s *StateStorage) Append(io io.Writer, mutation proto.Message, timeStamp time.Time) error {
+	typeName, ok := s.registry.GetTypeName(mutation)
+	if !ok {
+		return ErrUnknownType
+	}
+	event := &StorageEvent{
+		Type:      typeName,
+		TimeStamp: timeStamp,
+		Mutation:  mutation,
+	}
+	jsonBytes, err := json.Marshal(event)
+	if err != nil {
+		return err
+	}
+	if _, err := io.Write(jsonBytes); err != nil {
+		return err
+	}
+	io.Write([]byte("\n"))
+	return nil
+}
+
+func (s *StateStorage) Read(r *bufio.Scanner) (*StorageEvent, error) {
+	var event rawEvent
+
+	if r.Scan() {
+		b := r.Bytes()
+		err := json.Unmarshal(b, &event)
+		if err != nil {
+			return nil, err
+		}
+		typeName := event.Type
+		mutation, ok := s.registry.Create(typeName)
+		if !ok {
+			return nil, ErrUnknownType
+		}
+		if err := json.Unmarshal(event.Mutation, mutation); err != nil {
+			return nil, err
+		}
+		return &StorageEvent{
+			Type:      typeName,
+			TimeStamp: event.TimeStamp,
+			Mutation:  mutation,
+		}, r.Err()
+	}
+	return nil, io.EOF
+}