major changes

pkg/actor/base62-id.go

@@ -0,0 +1,131 @@
+package actor
+
+import (
+	"crypto/rand"
+	"encoding/json"
+	"fmt"
+)
+
+type GrainId uint64
+
+const base62Alphabet = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+
+// Reverse lookup (0xFF marks invalid)
+var base62Rev [256]byte
+
+func init() {
+	for i := range base62Rev {
+		base62Rev[i] = 0xFF
+	}
+	for i := 0; i < len(base62Alphabet); i++ {
+		base62Rev[base62Alphabet[i]] = byte(i)
+	}
+}
+
+// String returns the canonical base62 encoding of the 64-bit id.
+func (id GrainId) String() string {
+	return encodeBase62(uint64(id))
+}
+
+// MarshalJSON encodes the cart id as a JSON string.
+func (id GrainId) MarshalJSON() ([]byte, error) {
+	return json.Marshal(id.String())
+}
+
+// UnmarshalJSON decodes a cart id from a JSON string containing base62 text.
+func (id *GrainId) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return err
+	}
+	parsed, ok := ParseGrainId(s)
+	if !ok {
+		return fmt.Errorf("invalid cart id: %q", s)
+	}
+	*id = parsed
+	return nil
+}
+
+// NewGrainId generates a new cryptographically random non-zero 64-bit id.
+func NewGrainId() (GrainId, error) {
+	var b [8]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return 0, fmt.Errorf("NewGrainId: %w", err)
+	}
+	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; regenerate once to avoid "0" identifier if desired.
+		return NewGrainId()
+	}
+	return GrainId(u), nil
+}
+
+// MustNewGrainId panics if generation fails.
+func MustNewGrainId() GrainId {
+	id, err := NewGrainId()
+	if err != nil {
+		panic(err)
+	}
+	return id
+}
+
+// ParseGrainId parses a base62 string into a GrainId.
+// Returns (0,false) for invalid input.
+func ParseGrainId(s string) (GrainId, 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 0, false
+	}
+	return GrainId(u), true
+}
+
+// MustParseGrainId panics on invalid base62 input.
+func MustParseGrainId(s string) GrainId {
+	id, ok := ParseGrainId(s)
+	if !ok {
+		panic(fmt.Sprintf("invalid cart id: %q", s))
+	}
+	return id
+}
+
+// encodeBase62 converts a uint64 to base62 (shortest form).
+func encodeBase62(u uint64) string {
+	if u == 0 {
+		return "0"
+	}
+	var buf [11]byte
+	i := len(buf)
+	for u > 0 {
+		i--
+		buf[i] = base62Alphabet[u%62]
+		u /= 62
+	}
+	return string(buf[i:])
+}
+
+// decodeBase62 converts base62 text to uint64.
+func decodeBase62(s string) (uint64, bool) {
+	var v uint64
+	for i := 0; i < len(s); i++ {
+		c := s[i]
+		d := base62Rev[c]
+		if d == 0xFF {
+			return 0, false
+		}
+		v = v*62 + uint64(d)
+	}
+	return v, true
+}

pkg/actor/grain_pool.go

@@ -15,29 +15,29 @@ type MutationResult[V any] struct {
 
 type GrainPool[V any] interface {
 	Apply(ctx context.Context, id uint64, mutation ...proto.Message) (*MutationResult[V], error)
-	Get(ctx context.Context, id uint64) (V, error)
-	OwnerHost(id uint64) (Host, bool)
+	Get(ctx context.Context, id uint64) (*V, error)
+	OwnerHost(id uint64) (Host[V], bool)
 	Hostname() string
 	TakeOwnership(id uint64)
 	HandleOwnershipChange(host string, ids []uint64) error
 	HandleRemoteExpiry(host string, ids []uint64) error
 	Negotiate(otherHosts []string)
 	GetLocalIds() []uint64
+	IsHealthy() bool
+	Close()
+	IsKnown(string) bool
 	RemoveHost(host string)
 	AddRemoteHost(host string)
-	IsHealthy() bool
-	IsKnown(string) bool
-	Close()
 }
 
 // Host abstracts a remote node capable of proxying cart requests.
-type Host interface {
+type Host[V any] interface {
 	AnnounceExpiry(ids []uint64)
 	Negotiate(otherHosts []string) ([]string, error)
 	Name() string
 	Proxy(id uint64, w http.ResponseWriter, r *http.Request, customBody io.Reader) (bool, error)
-	Apply(ctx context.Context, id uint64, mutation ...proto.Message) (bool, error)
-	Get(ctx context.Context, id uint64, grain any) error
+	Apply(ctx context.Context, id uint64, mutation ...proto.Message) (*MutationResult[V], error)
+	Get(ctx context.Context, id uint64) (*V, error)
 	GetActorIds() []uint64
 	Close() error
 	Ping() bool

pkg/actor/grpc_server.go

@@ -107,19 +107,27 @@ func (s *ControlServer[V]) AnnounceOwnership(ctx context.Context, req *messages.
 	}, nil
 }
 
+func toAny[V any](grain V) (*anypb.Any, error) {
+	data, err := json.Marshal(grain)
+	if err != nil {
+		return nil, err
+	}
+	return &anypb.Any{
+		Value: data,
+	}, nil
+}
+
 func (s *ControlServer[V]) Get(ctx context.Context, req *messages.GetRequest) (*messages.GetReply, error) {
 	grain, err := s.pool.Get(ctx, req.Id)
 	if err != nil {
 		return nil, err
 	}
-	data, err := json.Marshal(grain)
+	grainAny, err := toAny(grain)
 	if err != nil {
 		return nil, err
 	}
 	return &messages.GetReply{
-		Grain: &anypb.Any{
-			Value: data,
-		},
+		Grain: grainAny,
 	}, nil
 }
 
@@ -163,16 +171,40 @@ func (s *ControlServer[V]) Apply(ctx context.Context, in *messages.ApplyRequest)
 	for i, anyMsg := range in.Messages {
 		msg, err := anyMsg.UnmarshalNew()
 		if err != nil {
-			return &messages.ApplyResult{Accepted: false}, fmt.Errorf("failed to unmarshal message: %w", err)
+			return nil, fmt.Errorf("failed to unmarshal message: %w", err)
 		}
 		msgs[i] = msg
 	}
-	_, err := s.pool.Apply(ctx, in.Id, msgs...)
+	r, err := s.pool.Apply(ctx, in.Id, msgs...)
 	if err != nil {
-		return &messages.ApplyResult{Accepted: false}, err
+		return nil, err
+	}
+	grainAny, err := toAny(r)
+	if err != nil {
+		return nil, err
+	}
+	mutList := make([]*messages.MutationResult, len(in.Messages))
+	for i, msg := range r.Mutations {
+		mut, err := anypb.New(msg.Mutation)
+		if err != nil {
+			return nil, err
+		}
+		var errString *string
+		if msg.Error != nil {
+			s := msg.Error.Error()
+			errString = &s
+		}
+		mutList[i] = &messages.MutationResult{
+			Type:    msg.Type,
+			Message: mut,
+			Error:   errString,
+		}
 	}
 
-	return &messages.ApplyResult{Accepted: true}, nil
+	return &messages.ApplyResult{
+		State:     grainAny,
+		Mutations: mutList,
+	}, nil
 }
 
 // ControlPlane: Negotiate (merge host views)

pkg/actor/grpc_server_test.go

@@ -17,14 +17,14 @@ type mockGrainPool struct {
 	applied []proto.Message
 }
 
-func (m *mockGrainPool) Apply(ctx context.Context, id uint64, mutations ...proto.Message) (*MutationResult[*mockGrain], error) {
+func (m *mockGrainPool) Apply(ctx context.Context, id uint64, mutations ...proto.Message) (*MutationResult[mockGrain], error) {
 	m.applied = mutations
 	// Simulate successful application
-	return &MutationResult[*mockGrain]{
-		Result: &mockGrain{},
+	return &MutationResult[mockGrain]{
+		Result: mockGrain{},
 		Mutations: []ApplyResult{
-			{Error: nil}, // Assume success
-			{Error: nil},
+			{Type: "AddItem", Mutation: &cart_messages.AddItem{ItemId: 1, Quantity: 2}, Error: nil},
+			{Type: "RemoveItem", Mutation: &cart_messages.RemoveItem{Id: 1}, Error: nil},
 		},
 	}, nil
 }
@@ -33,7 +33,7 @@ func (m *mockGrainPool) Get(ctx context.Context, id uint64) (*mockGrain, error)
 	return &mockGrain{}, nil
 }
 
-func (m *mockGrainPool) OwnerHost(id uint64) (Host, bool) {
+func (m *mockGrainPool) OwnerHost(id uint64) (Host[mockGrain], bool) {
 	return nil, false
 }
 
@@ -58,7 +58,7 @@ func TestApplyRequestWithMutations(t *testing.T) {
 	pool := &mockGrainPool{}
 
 	// Create gRPC server
-	server, err := NewControlServer[*mockGrain](DefaultServerConfig(), pool)
+	server, err := NewControlServer[mockGrain](DefaultServerConfig(), pool)
 	if err != nil {
 		t.Fatalf("failed to create server: %v", err)
 	}
@@ -88,8 +88,16 @@ func TestApplyRequestWithMutations(t *testing.T) {
 	}
 
 	// Verify response
-	if !resp.Accepted {
-		t.Errorf("expected Accepted=true, got false")
+	if resp.State == nil {
+		t.Errorf("expected State to be non-nil")
+	}
+	if len(resp.Mutations) != 2 {
+		t.Errorf("expected 2 mutation results, got %d", len(resp.Mutations))
+	}
+	for i, mut := range resp.Mutations {
+		if mut.Error != nil {
+			t.Errorf("expected no error in mutation %d, got %s", i, *mut.Error)
+		}
 	}
 
 	// Verify mutations were extracted and applied
@@ -103,3 +111,40 @@ func TestApplyRequestWithMutations(t *testing.T) {
 		t.Errorf("expected RemoveItem with Id=1, got %v", pool.applied[1])
 	}
 }
+
+func TestGetRequest(t *testing.T) {
+	// Setup mock pool
+	pool := &mockGrainPool{}
+
+	// Create gRPC server
+	server, err := NewControlServer[mockGrain](DefaultServerConfig(), pool)
+	if err != nil {
+		t.Fatalf("failed to create server: %v", err)
+	}
+	defer server.GracefulStop()
+
+	// Create client connection
+	conn, err := grpc.Dial("localhost:1337", grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("failed to dial: %v", err)
+	}
+	defer conn.Close()
+
+	client := control_plane_messages.NewControlPlaneClient(conn)
+
+	// Prepare GetRequest
+	req := &control_plane_messages.GetRequest{
+		Id: 123,
+	}
+
+	// Call Get
+	resp, err := client.Get(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Get failed: %v", err)
+	}
+
+	// Verify response
+	if resp.Grain == nil {
+		t.Errorf("expected Grain to be non-nil")
+	}
+}

pkg/actor/simple_grain_pool.go

@@ -18,7 +18,7 @@ type SimpleGrainPool[V any] struct {
 	mutationRegistry MutationRegistry
 	spawn            func(ctx context.Context, id uint64) (Grain[V], error)
 	destroy          func(grain Grain[V]) error
-	spawnHost        func(host string) (Host, error)
+	spawnHost        func(host string) (Host[V], error)
 	listeners        []LogListener
 	storage          LogStorage[V]
 	ttl              time.Duration
@@ -27,8 +27,8 @@ type SimpleGrainPool[V any] struct {
 	// Cluster coordination --------------------------------------------------
 	hostname     string
 	remoteMu     sync.RWMutex
-	remoteOwners map[uint64]Host
-	remoteHosts  map[string]Host
+	remoteOwners map[uint64]Host[V]
+	remoteHosts  map[string]Host[V]
 	//discardedHostHandler *DiscardedHostHandler
 
 	// House-keeping ---------------------------------------------------------
@@ -38,7 +38,7 @@ type SimpleGrainPool[V any] struct {
 type GrainPoolConfig[V any] struct {
 	Hostname         string
 	Spawn            func(ctx context.Context, id uint64) (Grain[V], error)
-	SpawnHost        func(host string) (Host, error)
+	SpawnHost        func(host string) (Host[V], error)
 	Destroy          func(grain Grain[V]) error
 	TTL              time.Duration
 	PoolSize         int
@@ -57,8 +57,8 @@ func NewSimpleGrainPool[V any](config GrainPoolConfig[V]) (*SimpleGrainPool[V],
 		ttl:              config.TTL,
 		poolSize:         config.PoolSize,
 		hostname:         config.Hostname,
-		remoteOwners:     make(map[uint64]Host),
-		remoteHosts:      make(map[string]Host),
+		remoteOwners:     make(map[uint64]Host[V]),
+		remoteHosts:      make(map[string]Host[V]),
 	}
 
 	p.purgeTicker = time.NewTicker(time.Minute)
@@ -99,7 +99,7 @@ func (p *SimpleGrainPool[V]) purge() {
 		}
 	}
 	p.localMu.Unlock()
-	p.forAllHosts(func(remote Host) {
+	p.forAllHosts(func(remote Host[V]) {
 		remote.AnnounceExpiry(purgedIds)
 	})
 
@@ -136,7 +136,6 @@ func (p *SimpleGrainPool[V]) HandleRemoteExpiry(host string, ids []uint64) error
 }
 
 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()
@@ -168,7 +167,7 @@ func (p *SimpleGrainPool[V]) AddRemoteHost(host string) {
 	p.AddRemote(host)
 }
 
-func (p *SimpleGrainPool[V]) AddRemote(host string) (Host, error) {
+func (p *SimpleGrainPool[V]) AddRemote(host string) (Host[V], error) {
 	if host == "" {
 		return nil, fmt.Errorf("host is empty")
 	}
@@ -200,7 +199,7 @@ func (p *SimpleGrainPool[V]) AddRemote(host string) (Host, error) {
 	return remote, nil
 }
 
-func (p *SimpleGrainPool[V]) initializeRemote(remote Host) {
+func (p *SimpleGrainPool[V]) initializeRemote(remote Host[V]) {
 
 	remotesIds := remote.GetActorIds()
 
@@ -268,7 +267,7 @@ func (p *SimpleGrainPool[V]) IsKnown(host string) bool {
 	return ok
 }
 
-func (p *SimpleGrainPool[V]) pingLoop(remote Host) {
+func (p *SimpleGrainPool[V]) pingLoop(remote Host[V]) {
 	remote.Ping()
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()
@@ -316,14 +315,14 @@ func (p *SimpleGrainPool[V]) SendNegotiation() {
 	p.remoteMu.RLock()
 	hosts := make([]string, 0, len(p.remoteHosts)+1)
 	hosts = append(hosts, p.hostname)
-	remotes := make([]Host, 0, len(p.remoteHosts))
+	remotes := make([]Host[V], 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) {
+	p.forAllHosts(func(remote Host[V]) {
 		knownByRemote, err := remote.Negotiate(hosts)
 
 		if err != nil {
@@ -338,7 +337,7 @@ func (p *SimpleGrainPool[V]) SendNegotiation() {
 	})
 }
 
-func (p *SimpleGrainPool[V]) forAllHosts(fn func(Host)) {
+func (p *SimpleGrainPool[V]) forAllHosts(fn func(Host[V])) {
 	p.remoteMu.RLock()
 	rh := maps.Clone(p.remoteHosts)
 	p.remoteMu.RUnlock()
@@ -366,7 +365,7 @@ func (p *SimpleGrainPool[V]) broadcastOwnership(ids []uint64) {
 		return
 	}
 
-	p.forAllHosts(func(rh Host) {
+	p.forAllHosts(func(rh Host[V]) {
 		rh.AnnounceOwnership(p.hostname, ids)
 	})
 	log.Printf("%s taking ownership of %d ids", p.hostname, len(ids))
@@ -385,10 +384,11 @@ func (p *SimpleGrainPool[V]) getOrClaimGrain(ctx context.Context, id uint64) (Gr
 	if err != nil {
 		return nil, err
 	}
+	go p.broadcastOwnership([]uint64{id})
 	p.localMu.Lock()
 	p.grains[id] = grain
 	p.localMu.Unlock()
-	go p.broadcastOwnership([]uint64{id})
+
 	return grain, nil
 }
 
@@ -396,7 +396,7 @@ func (p *SimpleGrainPool[V]) getOrClaimGrain(ctx context.Context, id uint64) (Gr
 // var ErrNotOwner = fmt.Errorf("not owner")
 
 // Apply applies a mutation to a grain.
-func (p *SimpleGrainPool[V]) Apply(ctx context.Context, id uint64, mutation ...proto.Message) (*MutationResult[*V], error) {
+func (p *SimpleGrainPool[V]) Apply(ctx context.Context, id uint64, mutation ...proto.Message) (*MutationResult[V], error) {
 	grain, err := p.getOrClaimGrain(ctx, id)
 	if err != nil {
 		return nil, err
@@ -421,8 +421,8 @@ func (p *SimpleGrainPool[V]) Apply(ctx context.Context, id uint64, mutation ...p
 		return nil, err
 	}
 
-	return &MutationResult[*V]{
-		Result:    result,
+	return &MutationResult[V]{
+		Result:    *result,
 		Mutations: mutations,
 	}, nil
 }
@@ -437,7 +437,7 @@ func (p *SimpleGrainPool[V]) Get(ctx context.Context, id uint64) (*V, error) {
 }
 
 // OwnerHost reports the remote owner (if any) for the supplied cart id.
-func (p *SimpleGrainPool[V]) OwnerHost(id uint64) (Host, bool) {
+func (p *SimpleGrainPool[V]) OwnerHost(id uint64) (Host[V], bool) {
 	p.remoteMu.RLock()
 	defer p.remoteMu.RUnlock()
 	owner, ok := p.remoteOwners[id]
@@ -452,7 +452,7 @@ func (p *SimpleGrainPool[V]) Hostname() string {
 // Close notifies remotes that this host is shutting down.
 func (p *SimpleGrainPool[V]) Close() {
 
-	p.forAllHosts(func(rh Host) {
+	p.forAllHosts(func(rh Host[V]) {
 		rh.Close()
 	})