mats/go-cart-actor
1
0
Fork 0
Code Issues Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

more otel
All checks were successful
Build and Publish / Metadata (push) Successful in 11s
Details
Build and Publish / BuildAndDeployAmd64 (push) Successful in 53s
Details
Build and Publish / BuildAndDeployArm64 (push) Successful in 4m49s
Details

Browse Source
This commit is contained in:
Mats Törnberg
2025-11-13 21:40:20 +01:00
parent cebd3ea80f
commit af5d4cd325
10 changed files with 98 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/backoffice/fileserver.go
View File

@@ -246,7 +246,7 @@ func (fs *FileServer) CartHandler(w http.ResponseWriter, r *http.Request) {
// reconstruct state from event log if present // reconstruct state from event log if present
grain := cart.NewCartGrain(id, time.Now()) grain := cart.NewCartGrain(id, time.Now())
err := fs.storage.LoadEvents(id, grain) err := fs.storage.LoadEvents(r.Context(), id, grain)
if err != nil { if err != nil {
writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()}) writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
return return

13
cmd/cart/checkout_server.go
View File

@@ -1,6 +1,7 @@
package main package main
import ( import (
"context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"log" "log"
@@ -27,12 +28,12 @@ var tpl = `<!DOCTYPE html>
</html> </html>
` `
func (a *App) getGrainFromOrder(order *CheckoutOrder) (*cart.CartGrain, error) { func (a *App) getGrainFromOrder(ctx context.Context, order *CheckoutOrder) (*cart.CartGrain, error) {
cartId, ok := cart.ParseCartId(order.MerchantReference1) cartId, ok := cart.ParseCartId(order.MerchantReference1)
if !ok { if !ok {
return nil, fmt.Errorf("invalid cart id in order reference: %s", order.MerchantReference1) return nil, fmt.Errorf("invalid cart id in order reference: %s", order.MerchantReference1)
} }
grain, err := a.pool.Get(uint64(cartId)) grain, err := a.pool.Get(ctx, uint64(cartId))
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to get cart grain: %w", err) return nil, fmt.Errorf("failed to get cart grain: %w", err)
} }
@@ -73,14 +74,14 @@ func (a *App) HandleCheckoutRequests(amqpUrl string, mux *http.ServeMux, invento
return return
} }
err = confirmOrder(order, orderHandler) err = confirmOrder(r.Context(), order, orderHandler)
if err != nil { if err != nil {
log.Printf("Error confirming order: %v\n", err) log.Printf("Error confirming order: %v\n", err)
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)
return return
} }
err = triggerOrderCompleted(a.server, order) err = triggerOrderCompleted(r.Context(), a.server, order)
if err != nil { if err != nil {
log.Printf("Error processing cart message: %v\n", err) log.Printf("Error processing cart message: %v\n", err)
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)
@@ -143,7 +144,7 @@ func (a *App) HandleCheckoutRequests(amqpUrl string, mux *http.ServeMux, invento
} }
log.Printf("Klarna order notification: %s", order.ID) log.Printf("Klarna order notification: %s", order.ID)
logger.InfoContext(r.Context(), "Klarna order notification received", "order_id", order.ID) logger.InfoContext(r.Context(), "Klarna order notification received", "order_id", order.ID)
grain, err := a.getGrainFromOrder(order) grain, err := a.getGrainFromOrder(r.Context(), order)
if err != nil { if err != nil {
logger.ErrorContext(r.Context(), "Unable to get grain from klarna order", "error", err.Error()) logger.ErrorContext(r.Context(), "Unable to get grain from klarna order", "error", err.Error())
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)
@@ -173,7 +174,7 @@ func (a *App) HandleCheckoutRequests(amqpUrl string, mux *http.ServeMux, invento
w.WriteHeader(http.StatusBadRequest) w.WriteHeader(http.StatusBadRequest)
} }
logger.InfoContext(r.Context(), "Klarna order validation received", "order_id", order.ID, "cart_id", order.MerchantReference1) logger.InfoContext(r.Context(), "Klarna order validation received", "order_id", order.ID, "cart_id", order.MerchantReference1)
grain, err := a.getGrainFromOrder(order) grain, err := a.getGrainFromOrder(r.Context(), order)
if err != nil { if err != nil {
logger.ErrorContext(r.Context(), "Unable to get grain from klarna order", "error", err.Error()) logger.ErrorContext(r.Context(), "Unable to get grain from klarna order", "error", err.Error())
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)

21
cmd/cart/main.go
View File

@@ -84,10 +84,13 @@ func main() {
reg := cart.NewCartMultationRegistry() reg := cart.NewCartMultationRegistry()
reg.RegisterProcessor( reg.RegisterProcessor(
actor.NewMutationProcessor(func(g *cart.CartGrain) error { actor.NewMutationProcessor(func(ctx context.Context, g *cart.CartGrain) error {
_, span := tracer.Start(ctx, "Totals and promotions")
defer span.End()
g.UpdateTotals() g.UpdateTotals()
ctx := promotions.NewContextFromCart(g, promotions.WithNow(time.Now()), promotions.WithCustomerSegment("vip"))
_, actions := promotionService.EvaluateAll(promotionData.State.Promotions, ctx) promotionCtx := promotions.NewContextFromCart(g, promotions.WithNow(time.Now()), promotions.WithCustomerSegment("vip"))
_, actions := promotionService.EvaluateAll(promotionData.State.Promotions, promotionCtx)
for _, action := range actions { for _, action := range actions {
log.Printf("apply: %+v", action) log.Printf("apply: %+v", action)
g.UpdateTotals() g.UpdateTotals()
@@ -100,12 +103,14 @@ func main() {
poolConfig := actor.GrainPoolConfig[cart.CartGrain]{ poolConfig := actor.GrainPoolConfig[cart.CartGrain]{
MutationRegistry: reg, MutationRegistry: reg,
Storage: diskStorage, Storage: diskStorage,
Spawn: func(id uint64) (actor.Grain[cart.CartGrain], error) { Spawn: func(ctx context.Context, id uint64) (actor.Grain[cart.CartGrain], error) {
_, span := tracer.Start(ctx, fmt.Sprintf("Spawn cart id %d", id))
defer span.End()
grainSpawns.Inc() grainSpawns.Inc()
ret := cart.NewCartGrain(id, time.Now()) ret := cart.NewCartGrain(id, time.Now())
// Set baseline lastChange at spawn; replay may update it to last event timestamp. // Set baseline lastChange at spawn; replay may update it to last event timestamp.
err := diskStorage.LoadEvents(id, ret) err := diskStorage.LoadEvents(ctx, id, ret)
return ret, err return ret, err
}, },
@@ -249,7 +254,7 @@ func main() {
} }
func triggerOrderCompleted(syncedServer *PoolServer, order *CheckoutOrder) error { func triggerOrderCompleted(ctx context.Context, syncedServer *PoolServer, order *CheckoutOrder) error {
mutation := &messages.OrderCreated{ mutation := &messages.OrderCreated{
OrderId: order.ID, OrderId: order.ID,
Status: order.Status, Status: order.Status,
@@ -258,12 +263,12 @@ func triggerOrderCompleted(syncedServer *PoolServer, order *CheckoutOrder) error
if !ok { if !ok {
return fmt.Errorf("invalid cart id in order reference: %s", order.MerchantReference1) return fmt.Errorf("invalid cart id in order reference: %s", order.MerchantReference1)
} }
_, applyErr := syncedServer.Apply(uint64(cid), mutation) _, applyErr := syncedServer.Apply(ctx, uint64(cid), mutation)
return applyErr return applyErr
} }
func confirmOrder(order *CheckoutOrder, orderHandler *AmqpOrderHandler) error { func confirmOrder(ctx context.Context, order *CheckoutOrder, orderHandler *AmqpOrderHandler) error {
orderToSend, err := json.Marshal(order) orderToSend, err := json.Marshal(order)
if err != nil { if err != nil {
return err return err

40
cmd/cart/pool-server.go
View File

@@ -55,12 +55,12 @@ func NewPoolServer(pool actor.GrainPool[*cart.CartGrain], pod_name string, klarn
} }
} }
func (s *PoolServer) ApplyLocal(id cart.CartId, mutation ...proto.Message) (*actor.MutationResult[*cart.CartGrain], error) { func (s *PoolServer) ApplyLocal(ctx context.Context, id cart.CartId, mutation ...proto.Message) (*actor.MutationResult[*cart.CartGrain], error) {
return s.Apply(uint64(id), mutation...) return s.Apply(ctx, uint64(id), mutation...)
} }
func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error { func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
grain, err := s.Get(uint64(id)) grain, err := s.Get(r.Context(), uint64(id))
if err != nil { if err != nil {
return err return err
} }
@@ -74,7 +74,7 @@ func (s *PoolServer) AddSkuToCartHandler(w http.ResponseWriter, r *http.Request,
if err != nil { if err != nil {
return err return err
} }
data, err := s.ApplyLocal(id, msg) data, err := s.ApplyLocal(r.Context(), id, msg)
if err != nil { if err != nil {
return err return err
} }
@@ -106,7 +106,7 @@ func (s *PoolServer) DeleteItemHandler(w http.ResponseWriter, r *http.Request, i
if err != nil { if err != nil {
return err return err
} }
data, err := s.ApplyLocal(id, &messages.RemoveItem{Id: uint32(itemId)}) data, err := s.ApplyLocal(r.Context(), id, &messages.RemoveItem{Id: uint32(itemId)})
if err != nil { if err != nil {
return err return err
} }
@@ -126,7 +126,7 @@ func (s *PoolServer) SetDeliveryHandler(w http.ResponseWriter, r *http.Request,
if err != nil { if err != nil {
return err return err
} }
data, err := s.ApplyLocal(id, &messages.SetDelivery{ data, err := s.ApplyLocal(r.Context(), id, &messages.SetDelivery{
Provider: delivery.Provider, Provider: delivery.Provider,
Items: delivery.Items, Items: delivery.Items,
PickupPoint: delivery.PickupPoint, PickupPoint: delivery.PickupPoint,
@@ -149,7 +149,7 @@ func (s *PoolServer) SetPickupPointHandler(w http.ResponseWriter, r *http.Reques
if err != nil { if err != nil {
return err return err
} }
reply, err := s.ApplyLocal(id, &messages.SetPickupPoint{ reply, err := s.ApplyLocal(r.Context(), id, &messages.SetPickupPoint{
DeliveryId: uint32(deliveryId), DeliveryId: uint32(deliveryId),
Id: pickupPoint.Id, Id: pickupPoint.Id,
Name: pickupPoint.Name, Name: pickupPoint.Name,
@@ -171,7 +171,7 @@ func (s *PoolServer) RemoveDeliveryHandler(w http.ResponseWriter, r *http.Reques
if err != nil { if err != nil {
return err return err
} }
reply, err := s.ApplyLocal(id, &messages.RemoveDelivery{Id: uint32(deliveryId)}) reply, err := s.ApplyLocal(r.Context(), id, &messages.RemoveDelivery{Id: uint32(deliveryId)})
if err != nil { if err != nil {
return err return err
} }
@@ -184,7 +184,7 @@ func (s *PoolServer) QuantityChangeHandler(w http.ResponseWriter, r *http.Reques
if err != nil { if err != nil {
return err return err
} }
reply, err := s.ApplyLocal(id, &changeQuantity) reply, err := s.ApplyLocal(r.Context(), id, &changeQuantity)
if err != nil { if err != nil {
return err return err
} }
@@ -234,7 +234,7 @@ func (s *PoolServer) SetCartItemsHandler(w http.ResponseWriter, r *http.Request,
msgs = append(msgs, &messages.ClearCartRequest{}) msgs = append(msgs, &messages.ClearCartRequest{})
msgs = append(msgs, getMultipleAddMessages(r.Context(), setCartItems.Items, setCartItems.Country)...) msgs = append(msgs, getMultipleAddMessages(r.Context(), setCartItems.Items, setCartItems.Country)...)
reply, err := s.ApplyLocal(id, msgs...) reply, err := s.ApplyLocal(r.Context(), id, msgs...)
if err != nil { if err != nil {
return err return err
} }
@@ -248,7 +248,7 @@ func (s *PoolServer) AddMultipleItemHandler(w http.ResponseWriter, r *http.Reque
return err return err
} }
reply, err := s.ApplyLocal(id, getMultipleAddMessages(r.Context(), setCartItems.Items, setCartItems.Country)...) reply, err := s.ApplyLocal(r.Context(), id, getMultipleAddMessages(r.Context(), setCartItems.Items, setCartItems.Country)...)
if err != nil { if err != nil {
return err return err
} }
@@ -272,7 +272,7 @@ func (s *PoolServer) AddSkuRequestHandler(w http.ResponseWriter, r *http.Request
if err != nil { if err != nil {
return err return err
} }
reply, err := s.ApplyLocal(id, msg) reply, err := s.ApplyLocal(r.Context(), id, msg)
if err != nil { if err != nil {
return err return err
} }
@@ -350,7 +350,7 @@ func (s *PoolServer) CreateOrUpdateCheckout(ctx context.Context, host string, id
} }
// Get current grain state (may be local or remote) // Get current grain state (may be local or remote)
grain, err := s.Get(uint64(id)) grain, err := s.Get(ctx, uint64(id))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -376,9 +376,9 @@ func (s *PoolServer) CreateOrUpdateCheckout(ctx context.Context, host string, id
} }
} }
func (s *PoolServer) ApplyCheckoutStarted(klarnaOrder *CheckoutOrder, id cart.CartId) (*actor.MutationResult[*cart.CartGrain], error) { func (s *PoolServer) ApplyCheckoutStarted(ctx context.Context, klarnaOrder *CheckoutOrder, id cart.CartId) (*actor.MutationResult[*cart.CartGrain], error) {
// Persist initialization state via mutation (best-effort) // Persist initialization state via mutation (best-effort)
return s.ApplyLocal(id, &messages.InitializeCheckout{ return s.ApplyLocal(ctx, id, &messages.InitializeCheckout{
OrderId: klarnaOrder.ID, OrderId: klarnaOrder.ID,
Status: klarnaOrder.Status, Status: klarnaOrder.Status,
PaymentInProgress: true, PaymentInProgress: true,
@@ -544,7 +544,7 @@ func (s *PoolServer) AddVoucherHandler(w http.ResponseWriter, r *http.Request, c
v := voucher.Service{} v := voucher.Service{}
msg, err := v.GetVoucher(data.VoucherCode) msg, err := v.GetVoucher(data.VoucherCode)
if err != nil { if err != nil {
s.ApplyLocal(cartId, &messages.PreConditionFailed{ s.ApplyLocal(r.Context(), cartId, &messages.PreConditionFailed{
Operation: "AddVoucher", Operation: "AddVoucher",
Error: err.Error(), Error: err.Error(),
}) })
@@ -552,7 +552,7 @@ func (s *PoolServer) AddVoucherHandler(w http.ResponseWriter, r *http.Request, c
w.Write([]byte(err.Error())) w.Write([]byte(err.Error()))
return err return err
} }
reply, err := s.ApplyLocal(cartId, msg) reply, err := s.ApplyLocal(r.Context(), cartId, msg)
if err != nil { if err != nil {
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error())) w.Write([]byte(err.Error()))
@@ -570,7 +570,7 @@ func (s *PoolServer) SubscriptionDetailsHandler(w http.ResponseWriter, r *http.R
w.Write([]byte(err.Error())) w.Write([]byte(err.Error()))
return err return err
} }
reply, err := s.ApplyLocal(cartId, data) reply, err := s.ApplyLocal(r.Context(), cartId, data)
if err != nil { if err != nil {
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error())) w.Write([]byte(err.Error()))
@@ -589,7 +589,7 @@ func (s *PoolServer) CheckoutHandler(fn func(order *CheckoutOrder, w http.Respon
logger.Error("unable to create klarna session", "error", err) logger.Error("unable to create klarna session", "error", err)
return err return err
} }
s.ApplyCheckoutStarted(order, cartId) s.ApplyCheckoutStarted(r.Context(), order, cartId)
return fn(order, w) return fn(order, w)
} }
order, err := s.klarnaClient.GetOrder(r.Context(), orderId) order, err := s.klarnaClient.GetOrder(r.Context(), orderId)
@@ -609,7 +609,7 @@ func (s *PoolServer) RemoveVoucherHandler(w http.ResponseWriter, r *http.Request
w.Write([]byte(err.Error())) w.Write([]byte(err.Error()))
return err return err
} }
reply, err := s.ApplyLocal(cartId, &messages.RemoveVoucher{Id: uint32(id)}) reply, err := s.ApplyLocal(r.Context(), cartId, &messages.RemoveVoucher{Id: uint32(id)})
if err != nil { if err != nil {
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error())) w.Write([]byte(err.Error()))

7
pkg/actor/disk_storage.go
View File

@@ -1,6 +1,7 @@
package actor package actor
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"log" "log"
@@ -25,7 +26,7 @@ type DiskStorage[V any] struct {
} }
type LogStorage[V any] interface { type LogStorage[V any] interface {
LoadEvents(id uint64, grain Grain[V]) error LoadEvents(ctx context.Context, id uint64, grain Grain[V]) error
AppendMutations(id uint64, msg ...proto.Message) error AppendMutations(id uint64, msg ...proto.Message) error
} }
@@ -86,7 +87,7 @@ func (s *DiskStorage[V]) logPath(id uint64) string {
return filepath.Join(s.path, fmt.Sprintf("%d.events.log", id)) return filepath.Join(s.path, fmt.Sprintf("%d.events.log", id))
} }
func (s *DiskStorage[V]) LoadEvents(id uint64, grain Grain[V]) error { func (s *DiskStorage[V]) LoadEvents(ctx context.Context, id uint64, grain Grain[V]) error {
path := s.logPath(id) path := s.logPath(id)
if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) { if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
// No log -> nothing to replay // No log -> nothing to replay
@@ -99,7 +100,7 @@ func (s *DiskStorage[V]) LoadEvents(id uint64, grain Grain[V]) error {
} }
defer fh.Close() defer fh.Close()
return s.Load(fh, func(msg proto.Message) { return s.Load(fh, func(msg proto.Message) {
s.registry.Apply(grain, msg) s.registry.Apply(ctx, grain, msg)
}) })
} }

5
pkg/actor/grain_pool.go
View File

@@ -1,6 +1,7 @@
package actor package actor
import ( import (
"context"
"net/http" "net/http"
"github.com/gogo/protobuf/proto" "github.com/gogo/protobuf/proto"
@@ -12,8 +13,8 @@ type MutationResult[V any] struct {
} }
type GrainPool[V any] interface { type GrainPool[V any] interface {
Apply(id uint64, mutation ...proto.Message) (*MutationResult[V], error) Apply(ctx context.Context, id uint64, mutation ...proto.Message) (*MutationResult[V], error)
Get(id uint64) (V, error) Get(ctx context.Context, id uint64) (V, error)
OwnerHost(id uint64) (Host, bool) OwnerHost(id uint64) (Host, bool)
Hostname() string Hostname() string
TakeOwnership(id uint64) TakeOwnership(id uint64)

44
pkg/actor/mutation_registry.go
View File

@@ -1,12 +1,14 @@
package actor package actor
import ( import (
"context"
"fmt" "fmt"
"log" "log"
"reflect" "reflect"
"sync" "sync"
"github.com/gogo/protobuf/proto" "github.com/gogo/protobuf/proto"
"go.opentelemetry.io/otel/attribute"
) )
type ApplyResult struct { type ApplyResult struct {
@@ -16,27 +18,25 @@ type ApplyResult struct {
} }
type MutationProcessor interface { type MutationProcessor interface {
Process(grain any) error Process(ctx context.Context, grain any) error
} }
type BasicMutationProcessor[V any] struct { type BasicMutationProcessor[V any] struct {
processor func(any) error processor func(ctx context.Context, grain V) error
} }
func NewMutationProcessor[V any](process func(V) error) MutationProcessor { func NewMutationProcessor[V any](process func(ctx context.Context, grain V) error) MutationProcessor {
return &BasicMutationProcessor[V]{ return &BasicMutationProcessor[V]{
processor: func(v any) error { processor: process,
return process(v.(V))
},
} }
} }
func (p *BasicMutationProcessor[V]) Process(grain any) error { func (p *BasicMutationProcessor[V]) Process(ctx context.Context, grain any) error {
return p.processor(grain) return p.processor(ctx, grain.(V))
} }
type MutationRegistry interface { type MutationRegistry interface {
Apply(grain any, msg ...proto.Message) ([]ApplyResult, error) Apply(ctx context.Context, grain any, msg ...proto.Message) ([]ApplyResult, error)
RegisterMutations(handlers ...MutationHandler) RegisterMutations(handlers ...MutationHandler)
Create(typeName string) (proto.Message, bool) Create(typeName string) (proto.Message, bool)
GetTypeName(msg proto.Message) (string, bool) GetTypeName(msg proto.Message) (string, bool)
@@ -192,7 +192,15 @@ func (r *ProtoMutationRegistry) Create(typeName string) (proto.Message, bool) {
// Returns updated grain if successful. // Returns updated grain if successful.
// //
// If the mutation is not registered, returns (nil, ErrMutationNotRegistered). // If the mutation is not registered, returns (nil, ErrMutationNotRegistered).
func (r *ProtoMutationRegistry) Apply(grain any, msg ...proto.Message) ([]ApplyResult, error) { func (r *ProtoMutationRegistry) Apply(ctx context.Context, grain any, msg ...proto.Message) ([]ApplyResult, error) {
parentCtx, span := tracer.Start(ctx, "apply mutations")
defer span.End()
span.SetAttributes(
attribute.String("component", "registry"),
attribute.Int("mutations", len(msg)),
)
results := make([]ApplyResult, 0, len(msg)) results := make([]ApplyResult, 0, len(msg))
if grain == nil { if grain == nil {
@@ -214,20 +222,30 @@ func (r *ProtoMutationRegistry) Apply(grain any, msg ...proto.Message) ([]ApplyR
continue continue
} }
rt := indirectType(reflect.TypeOf(m)) rt := indirectType(reflect.TypeOf(m))
_, msgSpan := tracer.Start(parentCtx, rt.Name())
r.mutationRegistryMu.RLock() r.mutationRegistryMu.RLock()
entry, ok := r.mutationRegistry[rt] entry, ok := r.mutationRegistry[rt]
r.mutationRegistryMu.RUnlock() r.mutationRegistryMu.RUnlock()
if !ok { if !ok {
results = append(results, ApplyResult{Error: ErrMutationNotRegistered, Type: rt.Name(), Mutation: m}) results = append(results, ApplyResult{Error: ErrMutationNotRegistered, Type: rt.Name(), Mutation: m})
continue continue
} else {
err := entry.Handle(grain, m)
if err != nil {
msgSpan.RecordError(err)
}
results = append(results, ApplyResult{Error: err, Type: rt.Name(), Mutation: m})
} }
err := entry.Handle(grain, m) msgSpan.End()
results = append(results, ApplyResult{Error: err, Type: rt.Name(), Mutation: m})
} }
if len(results) > 0 { if len(results) > 0 {
processCtx, processSpan := tracer.Start(ctx, "after mutation processors")
defer processSpan.End()
for _, processor := range r.processors { for _, processor := range r.processors {
err := processor.Process(grain)
err := processor.Process(processCtx, grain)
if err != nil { if err != nil {
return results, err return results, err
} }

7
pkg/actor/mutation_registry_test.go
View File

@@ -1,6 +1,7 @@
package actor package actor
import ( import (
"context"
"errors" "errors"
"reflect" "reflect"
"slices" "slices"
@@ -78,7 +79,7 @@ func TestRegisteredMutationBasics(t *testing.T) {
// Apply happy path // Apply happy path
state := &cartState{} state := &cartState{}
add := &messages.AddItem{ItemId: 42, Quantity: 3, Sku: "ABC"} add := &messages.AddItem{ItemId: 42, Quantity: 3, Sku: "ABC"}
if _, err := reg.Apply(state, add); err != nil { if _, err := reg.Apply(context.Background(), state, add); err != nil {
t.Fatalf("Apply returned error: %v", err) t.Fatalf("Apply returned error: %v", err)
} }
if state.calls != 1 { if state.calls != 1 {
@@ -94,12 +95,12 @@ func TestRegisteredMutationBasics(t *testing.T) {
} }
// Apply nil message // Apply nil message
if _, err := reg.Apply(state, nil); err == nil { if _, err := reg.Apply(context.Background(), state, nil); err == nil {
t.Fatalf("expected error for nil mutation message") t.Fatalf("expected error for nil mutation message")
} }
// Apply unregistered message // Apply unregistered message
if _, err := reg.Apply(state, &messages.Noop{}); !errors.Is(err, ErrMutationNotRegistered) { if _, err := reg.Apply(context.Background(), state, &messages.Noop{}); !errors.Is(err, ErrMutationNotRegistered) {
t.Fatalf("expected ErrMutationNotRegistered, got %v", err) t.Fatalf("expected ErrMutationNotRegistered, got %v", err)
} }
} }

19
pkg/actor/simple_grain_pool.go
View File

@@ -1,6 +1,7 @@
package actor package actor
import ( import (
"context"
"fmt" "fmt"
"log" "log"
"maps" "maps"
@@ -15,7 +16,7 @@ type SimpleGrainPool[V any] struct {
localMu sync.RWMutex localMu sync.RWMutex
grains map[uint64]Grain[V] grains map[uint64]Grain[V]
mutationRegistry MutationRegistry mutationRegistry MutationRegistry
spawn func(id uint64) (Grain[V], error) spawn func(ctx context.Context, id uint64) (Grain[V], error)
spawnHost func(host string) (Host, error) spawnHost func(host string) (Host, error)
listeners []LogListener listeners []LogListener
storage LogStorage[V] storage LogStorage[V]
@@ -35,7 +36,7 @@ type SimpleGrainPool[V any] struct {
type GrainPoolConfig[V any] struct { type GrainPoolConfig[V any] struct {
Hostname string Hostname string
Spawn func(id uint64) (Grain[V], error) Spawn func(ctx context.Context, id uint64) (Grain[V], error)
SpawnHost func(host string) (Host, error) SpawnHost func(host string) (Host, error)
TTL time.Duration TTL time.Duration
PoolSize int PoolSize int
@@ -366,7 +367,7 @@ func (p *SimpleGrainPool[V]) broadcastOwnership(ids []uint64) {
// go p.statsUpdate() // go p.statsUpdate()
} }
func (p *SimpleGrainPool[V]) getOrClaimGrain(id uint64) (Grain[V], error) { func (p *SimpleGrainPool[V]) getOrClaimGrain(ctx context.Context, id uint64) (Grain[V], error) {
p.localMu.RLock() p.localMu.RLock()
grain, exists := p.grains[id] grain, exists := p.grains[id]
p.localMu.RUnlock() p.localMu.RUnlock()
@@ -374,7 +375,7 @@ func (p *SimpleGrainPool[V]) getOrClaimGrain(id uint64) (Grain[V], error) {
return grain, nil return grain, nil
} }
grain, err := p.spawn(id) grain, err := p.spawn(ctx, id)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -389,13 +390,13 @@ func (p *SimpleGrainPool[V]) getOrClaimGrain(id uint64) (Grain[V], error) {
// var ErrNotOwner = fmt.Errorf("not owner") // var ErrNotOwner = fmt.Errorf("not owner")
// Apply applies a mutation to a grain. // Apply applies a mutation to a grain.
func (p *SimpleGrainPool[V]) Apply(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(id) grain, err := p.getOrClaimGrain(ctx, id)
if err != nil { if err != nil {
return nil, err return nil, err
} }
mutations, err := p.mutationRegistry.Apply(grain, mutation...) mutations, err := p.mutationRegistry.Apply(ctx, grain, mutation...)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -420,8 +421,8 @@ func (p *SimpleGrainPool[V]) Apply(id uint64, mutation ...proto.Message) (*Mutat
} }
// Get returns the current state of a grain. // Get returns the current state of a grain.
func (p *SimpleGrainPool[V]) Get(id uint64) (*V, error) { func (p *SimpleGrainPool[V]) Get(ctx context.Context, id uint64) (*V, error) {
grain, err := p.getOrClaimGrain(id) grain, err := p.getOrClaimGrain(ctx, id)
if err != nil { if err != nil {
return nil, err return nil, err
} }

9
pkg/cart/mutation_test.go
View File

@@ -1,6 +1,7 @@
package cart package cart
import ( import (
"context"
"reflect" "reflect"
"slices" "slices"
"strings" "strings"
@@ -90,7 +91,7 @@ func ptr[T any](v T) *T { return &v }
func applyOne(t *testing.T, reg actor.MutationRegistry, g *CartGrain, msg proto.Message) actor.ApplyResult { func applyOne(t *testing.T, reg actor.MutationRegistry, g *CartGrain, msg proto.Message) actor.ApplyResult {
t.Helper() t.Helper()
results, err := reg.Apply(g, msg) results, err := reg.Apply(context.Background(), g, msg)
if err != nil { if err != nil {
t.Fatalf("unexpected registry-level error applying %T: %v", msg, err) t.Fatalf("unexpected registry-level error applying %T: %v", msg, err)
} }
@@ -178,7 +179,7 @@ func TestMutationRegistryCoverage(t *testing.T) {
} }
// Apply unregistered message -> result should contain ErrMutationNotRegistered, no top-level error // Apply unregistered message -> result should contain ErrMutationNotRegistered, no top-level error
results, err := reg.Apply(newTestGrain(), &messages.Noop{}) results, err := reg.Apply(context.Background(), newTestGrain(), &messages.Noop{})
if err != nil { if err != nil {
t.Fatalf("unexpected top-level error applying unregistered mutation: %v", err) t.Fatalf("unexpected top-level error applying unregistered mutation: %v", err)
} }
@@ -426,7 +427,7 @@ func TestSubscriptionDetailsMutation(t *testing.T) {
applyErrorContains(t, reg, g, &messages.UpsertSubscriptionDetails{Id: &badId}, "not found") applyErrorContains(t, reg, g, &messages.UpsertSubscriptionDetails{Id: &badId}, "not found")
// Nil mutation should be ignored and produce zero results. // Nil mutation should be ignored and produce zero results.
resultsNil, errNil := reg.Apply(g, (*messages.UpsertSubscriptionDetails)(nil)) resultsNil, errNil := reg.Apply(context.Background(), g, (*messages.UpsertSubscriptionDetails)(nil))
if errNil != nil { if errNil != nil {
t.Fatalf("unexpected error for nil mutation element: %v", errNil) t.Fatalf("unexpected error for nil mutation element: %v", errNil)
} }
@@ -450,7 +451,7 @@ func TestRegistryDefensiveErrors(t *testing.T) {
} }
// Nil message slice // Nil message slice
results, _ = reg.Apply(g, nil) results, _ = reg.Apply(context.Background(), g, nil)
if len(results) != 0 { if len(results) != 0 {
t.Fatalf("expected no results when message slice nil") t.Fatalf("expected no results when message slice nil")