From 8e60cc2239c9aa36ead5d25c6f75da506be2976e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mats=20T=C3=B6rnberg?= Date: Mon, 13 Oct 2025 19:51:27 +0200 Subject: [PATCH] support multiple mutations --- cmd/cart/pool-server.go | 26 ++++++++++++++------------ pkg/actor/disk_storage.go | 14 +++++++++----- pkg/actor/grain_pool.go | 2 +- pkg/actor/mutation_registry.go | 11 ++++++----- pkg/actor/simple_grain_pool.go | 6 +++--- 5 files changed, 33 insertions(+), 26 deletions(-) diff --git a/cmd/cart/pool-server.go b/cmd/cart/pool-server.go index 6862255..047fe32 100644 --- a/cmd/cart/pool-server.go +++ b/cmd/cart/pool-server.go @@ -29,8 +29,8 @@ func NewPoolServer(pool actor.GrainPool[*CartGrain], pod_name string, klarnaClie } } -func (s *PoolServer) ApplyLocal(id CartId, mutation proto.Message) (*CartGrain, error) { - return s.Apply(uint64(id), mutation) +func (s *PoolServer) ApplyLocal(id CartId, mutation ...proto.Message) (*CartGrain, error) { + return s.Apply(uint64(id), mutation...) } func (s *PoolServer) HandleGet(w http.ResponseWriter, r *http.Request, id CartId) error { @@ -178,28 +178,30 @@ func (s *PoolServer) HandleSetCartItems(w http.ResponseWriter, r *http.Request, if err != nil { return err } - reply, err := s.ApplyLocal(id, &messages.ClearCartRequest{}) - if err != nil { - return err - } + + msgs := make([]proto.Message, 0, len(setCartItems.Items)+1) + msgs = append(msgs, &messages.ClearCartRequest{}) wg := sync.WaitGroup{} for _, item := range setCartItems.Items { wg.Add(1) go func(sku string, quantity int) { + defer wg.Done() msg, err := GetItemAddMessage(sku, quantity, setCartItems.Country, nil) if err != nil { log.Printf("error adding item %s: %v", sku, err) return } - reply, err = s.ApplyLocal(id, msg) - if err != nil { - log.Printf("error applying message %v: %v", msg, err) - return - } - wg.Done() + msgs = append(msgs, msg) + }(item.Sku, item.Quantity) } + wg.Wait() + + reply, err := s.ApplyLocal(id, msgs...) + if err != nil { + return err + } return s.WriteResult(w, reply) } diff --git a/pkg/actor/disk_storage.go b/pkg/actor/disk_storage.go index dc4c4dd..f82fbdd 100644 --- a/pkg/actor/disk_storage.go +++ b/pkg/actor/disk_storage.go @@ -26,7 +26,7 @@ type DiskStorage[V any] struct { type LogStorage[V any] interface { LoadEvents(id uint64, grain Grain[V]) error - AppendEvent(id uint64, msg proto.Message) error + AppendEvent(id uint64, msg ...proto.Message) error } func NewDiskStorage[V any](path string, registry MutationRegistry) *DiskStorage[V] { @@ -102,14 +102,16 @@ func (s *DiskStorage[V]) Close() { close(s.done) } -func (s *DiskStorage[V]) AppendEvent(id uint64, msg proto.Message) error { +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) } - queue = append(queue, QueueEvent{Message: msg}) + for _, m := range msg { + queue = append(queue, QueueEvent{Message: m, TimeStamp: time.Now()}) + } s.queue.Store(id, queue) return nil } else { @@ -120,8 +122,10 @@ func (s *DiskStorage[V]) AppendEvent(id uint64, msg proto.Message) error { return err } defer fh.Close() - - return s.Append(fh, msg, time.Now()) + for _, m := range msg { + err = s.Append(fh, m, time.Now()) + } + return err } } diff --git a/pkg/actor/grain_pool.go b/pkg/actor/grain_pool.go index 98366f7..d3dac8e 100644 --- a/pkg/actor/grain_pool.go +++ b/pkg/actor/grain_pool.go @@ -7,7 +7,7 @@ import ( ) type GrainPool[V any] interface { - Apply(id uint64, mutation proto.Message) (V, error) + Apply(id uint64, mutation ...proto.Message) (V, error) Get(id uint64) (V, error) OwnerHost(id uint64) (Host, bool) Hostname() string diff --git a/pkg/actor/mutation_registry.go b/pkg/actor/mutation_registry.go index e4f1172..9fdfa88 100644 --- a/pkg/actor/mutation_registry.go +++ b/pkg/actor/mutation_registry.go @@ -10,7 +10,7 @@ import ( ) type MutationRegistry interface { - Apply(grain any, msg proto.Message) error + Apply(grain any, msg ...proto.Message) error RegisterMutations(handlers ...MutationHandler) Create(typeName string) (proto.Message, bool) GetTypeName(msg proto.Message) (string, bool) @@ -145,7 +145,7 @@ func (r *ProtoMutationRegistry) Create(typeName string) (proto.Message, bool) { // Returns updated grain if successful. // // If the mutation is not registered, returns (nil, ErrMutationNotRegistered). -func (r *ProtoMutationRegistry) Apply(grain any, msg proto.Message) error { +func (r *ProtoMutationRegistry) Apply(grain any, msg ...proto.Message) error { if grain == nil { return fmt.Errorf("nil grain") } @@ -161,9 +161,10 @@ func (r *ProtoMutationRegistry) Apply(grain any, msg proto.Message) error { if !ok { return ErrMutationNotRegistered } - - if err := entry.Handle(grain, msg); err != nil { - return err + for _, m := range msg { + if err := entry.Handle(grain, m); err != nil { + return err + } } // if entry.updateTotals { diff --git a/pkg/actor/simple_grain_pool.go b/pkg/actor/simple_grain_pool.go index 05a500d..2722f86 100644 --- a/pkg/actor/simple_grain_pool.go +++ b/pkg/actor/simple_grain_pool.go @@ -362,17 +362,17 @@ func (p *SimpleGrainPool[V]) getOrClaimGrain(id uint64) (Grain[V], error) { // var ErrNotOwner = fmt.Errorf("not owner") // Apply applies a mutation to a grain. -func (p *SimpleGrainPool[V]) Apply(id uint64, mutation proto.Message) (*V, error) { +func (p *SimpleGrainPool[V]) Apply(id uint64, mutation ...proto.Message) (*V, error) { grain, err := p.getOrClaimGrain(id) if err != nil { return nil, err } - if applyErr := p.mutationRegistry.Apply(grain, mutation); applyErr != nil { + if applyErr := p.mutationRegistry.Apply(grain, mutation...); applyErr != nil { return nil, applyErr } if p.storage != nil { go func() { - if err := p.storage.AppendEvent(id, mutation); err != nil { + if err := p.storage.AppendEvent(id, mutation...); err != nil { log.Printf("failed to store mutation for grain %d: %v", id, err) } }()