Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions example/go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -154,8 +154,8 @@ require (
github.com/tidwall/btree v1.8.1 // indirect
github.com/tidwall/match v1.2.0 // indirect
github.com/tidwall/redcon v1.6.2 // indirect
github.com/tochemey/goakt/v4 v4.2.13 // indirect
github.com/tochemey/olric v0.3.10 // indirect
github.com/tochemey/goakt/v4 v4.3.1 // indirect
github.com/tochemey/olric v0.3.11 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
Expand Down
8 changes: 4 additions & 4 deletions example/go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -530,10 +530,10 @@ github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRU
github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI=
github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4=
github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg=
github.com/tochemey/goakt/v4 v4.2.13 h1:Pq7+EGApre5d8rUMFD/EcZmtsKnTQ4ALuNKQ1a1O6/Q=
github.com/tochemey/goakt/v4 v4.2.13/go.mod h1:KxXy+niDlSrdeLnWkNg+nDQ5BCZGxrkKZEAQbBSmxNY=
github.com/tochemey/olric v0.3.10 h1:S5Ypz0QxZiLFv7DUrm9q40+mh9FO24WniZRabU33aQE=
github.com/tochemey/olric v0.3.10/go.mod h1:l/LsIx3UPjIKh4OYOihM7+7a66hAmdKJ+fX6YJhfH7k=
github.com/tochemey/goakt/v4 v4.3.1 h1:l6JBN/kr1qN9tFki2dfDUDIwhmBnirxEp7h8CGZNyq8=
github.com/tochemey/goakt/v4 v4.3.1/go.mod h1:QcHNrLicr/4+dqzI0OEJ//8rv0TxpvbqHsEkWSuCcAc=
github.com/tochemey/olric v0.3.11 h1:GMvXyIc6eeBgFknHVLyVS1GMBQoi8HhNU/ml1cZ1Nm4=
github.com/tochemey/olric v0.3.11/go.mod h1:lauJDgUkt283EIMzO1mMh9wLLVQSwREtBpF/lYX3JKU=
github.com/ttacon/chalk v0.0.0-20160626202418-22c06c80ed31/go.mod h1:onvgF043R+lC5RZ8IT9rBXDaEDnpnw/Cl+HFiw+v/7Q=
github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
Expand Down
4 changes: 2 additions & 2 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ require (
github.com/stretchr/testify v1.11.1
github.com/stripe/stripe-go/v82 v82.5.1
github.com/tliron/glsp v0.2.2
github.com/tochemey/goakt/v4 v4.2.13
github.com/tochemey/goakt/v4 v4.3.1
github.com/xdg-go/scram v1.2.0
github.com/zalando/go-keyring v0.2.8
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0
Expand Down Expand Up @@ -228,7 +228,7 @@ require (
github.com/tidwall/redcon v1.6.2 // indirect
github.com/tliron/commonlog v0.2.21 // indirect
github.com/tliron/go-kutil v0.4.0 // indirect
github.com/tochemey/olric v0.3.10 // indirect
github.com/tochemey/olric v0.3.11 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
Expand Down
8 changes: 4 additions & 4 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -651,10 +651,10 @@ github.com/tliron/glsp v0.2.2 h1:IKPfwpE8Lu8yB6Dayta+IyRMAbTVunudeauEgjXBt+c=
github.com/tliron/glsp v0.2.2/go.mod h1:GMVWDNeODxHzmDPvYbYTCs7yHVaEATfYtXiYJ9w1nBg=
github.com/tliron/go-kutil v0.4.0 h1:5JwcBacgnqS3XyhwCWZKvq8ftlbVttNXnt+kfCH+Y2E=
github.com/tliron/go-kutil v0.4.0/go.mod h1:hpHVq+CP1uci2M208UEjPiPwsRsz/QweGBnLB3CaQ24=
github.com/tochemey/goakt/v4 v4.2.13 h1:Pq7+EGApre5d8rUMFD/EcZmtsKnTQ4ALuNKQ1a1O6/Q=
github.com/tochemey/goakt/v4 v4.2.13/go.mod h1:KxXy+niDlSrdeLnWkNg+nDQ5BCZGxrkKZEAQbBSmxNY=
github.com/tochemey/olric v0.3.10 h1:S5Ypz0QxZiLFv7DUrm9q40+mh9FO24WniZRabU33aQE=
github.com/tochemey/olric v0.3.10/go.mod h1:l/LsIx3UPjIKh4OYOihM7+7a66hAmdKJ+fX6YJhfH7k=
github.com/tochemey/goakt/v4 v4.3.1 h1:l6JBN/kr1qN9tFki2dfDUDIwhmBnirxEp7h8CGZNyq8=
github.com/tochemey/goakt/v4 v4.3.1/go.mod h1:QcHNrLicr/4+dqzI0OEJ//8rv0TxpvbqHsEkWSuCcAc=
github.com/tochemey/olric v0.3.11 h1:GMvXyIc6eeBgFknHVLyVS1GMBQoi8HhNU/ml1cZ1Nm4=
github.com/tochemey/olric v0.3.11/go.mod h1:lauJDgUkt283EIMzO1mMh9wLLVQSwREtBpF/lYX3JKU=
github.com/ttacon/chalk v0.0.0-20160626202418-22c06c80ed31/go.mod h1:onvgF043R+lC5RZ8IT9rBXDaEDnpnw/Cl+HFiw+v/7Q=
github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
Expand Down
20 changes: 19 additions & 1 deletion plugins/actors/bridge_actor.go
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,25 @@ type BridgeGrain struct {
}

// OnActivate initializes grain state when the grain is loaded into memory.
func (g *BridgeGrain) OnActivate(_ context.Context, _ *goaktactor.GrainProps) error {
func (g *BridgeGrain) OnActivate(_ context.Context, props *goaktactor.GrainProps) error {
var poolName string
for _, dep := range props.Dependencies() {
if ref, ok := dep.(*grainPoolReference); ok {
poolName = ref.poolName
break
}
}
registry, ok := props.ActorSystem().Extension(grainPoolRegistryID).(*grainPoolRegistry)
if !ok || poolName == "" {
return fmt.Errorf("actor grain: pool configuration is missing")
}
poolValue, ok := registry.pools.Load(poolName)
if !ok {
return fmt.Errorf("actor grain: pool %q not registered", poolName)
}
pool := poolValue.(*ActorPoolModule)
g.poolName, g.handlers = pool.name, pool.handlers
g.registry, g.app, g.logger = pool.stepRegistry, pool.app, pool.logger
if g.state == nil {
g.state = make(map[string]any)
}
Expand Down
110 changes: 96 additions & 14 deletions plugins/actors/bridge_actor_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,98 @@ package actors

import (
"context"
"fmt"
"strings"
"testing"
"time"

"github.com/GoCodeAlone/modular"
"github.com/GoCodeAlone/workflow/interfaces"
"github.com/GoCodeAlone/workflow/pipeline"
"github.com/tochemey/goakt/v4/actor"
)

func TestBridgeGrain_RequiresPoolConfiguration(t *testing.T) {
ctx := context.Background()
sys, err := actor.NewActorSystem("test-unconfigured-grain")
if err != nil {
t.Fatal(err)
}
if err := sys.Start(ctx); err != nil {
t.Fatal(err)
}
defer sys.Stop(ctx) //nolint:errcheck
_, err = actor.GrainOf[*BridgeGrain](ctx, sys, "unconfigured", actor.WithGrainInitMaxRetries(0))
if err == nil {
t.Fatal("zero-value grain activated without its pool configuration")
}
if !strings.Contains(err.Error(), "pool configuration is missing") {
t.Fatalf("unexpected activation failure: %v", err)
}
}

type grainTestRegistry struct {
app modular.Application
}

func (*grainTestRegistry) Types() []string { return []string{"step.grain-test"} }

func (r *grainTestRegistry) Create(stepType, name string, config map[string]any, app any) (interfaces.PipelineStep, error) {
if stepType != "step.grain-test" || app != r.app {
return nil, fmt.Errorf("grain did not receive the pool registry and application")
}
return pipeline.NewSetStepFactory()(name, config, r.app)
}

func TestBridgeGrain_PoolServicesSurviveDependencyRoundTrip(t *testing.T) {
ctx := context.Background()
sysMod, err := NewActorSystemModule("test-grain-resources", map[string]any{"shutdownTimeout": "5s"})
if err != nil {
t.Fatal(err)
}
if err := sysMod.Start(ctx); err != nil {
t.Fatal(err)
}
defer sysMod.Stop(ctx) //nolint:errcheck
app := modular.NewStdApplication(modular.NewStdConfigProvider(nil), nil)
for _, poolName := range []string{"orders", "payments"} {
pool := &ActorPoolModule{
name: poolName, system: sysMod, idleTimeout: time.Minute,
app: app, stepRegistry: &grainTestRegistry{app: app},
handlers: map[string]*HandlerPipeline{"Ping": {
Steps: []map[string]any{{"name": "result", "type": "step.grain-test", "config": map[string]any{
"values": map[string]any{"pool": "{{ .actor.pool }}"},
}}},
}},
}
if _, err := pool.GetGrainIdentity(ctx, poolName+"-initial"); err != nil {
t.Fatal(err)
}
encoded, err := (&grainPoolReference{poolName: poolName}).MarshalBinary()
if err != nil {
t.Fatal(err)
}
var decoded grainPoolReference
if err := decoded.UnmarshalBinary(encoded); err != nil {
t.Fatal(err)
}
// Activate a fresh zero-value grain with the reconstructed reference.
grainID, err := actor.GrainOf[*BridgeGrain](ctx, sysMod.ActorSystem(), poolName+"-recreated",
actor.WithGrainDependencies(&decoded), actor.WithGrainDeactivateAfter(time.Minute))
if err != nil {
t.Fatal(err)
}
resp, err := sysMod.ActorSystem().AskGrain(ctx, grainID, &ActorMessage{Type: "Ping"}, 5*time.Second)
if err != nil {
t.Fatal(err)
}
result, ok := resp.(map[string]any)
if !ok || result["pool"] != poolName {
t.Fatalf("pool %s: response %v, want matching pool configuration", poolName, resp)
}
}
}

func TestBridgeGrain_ReceiveAndStatePersistence(t *testing.T) {
ctx := context.Background()

Expand Down Expand Up @@ -40,27 +126,23 @@ func TestBridgeGrain_ReceiveAndStatePersistence(t *testing.T) {
},
}

sys, err := actor.NewActorSystem("test-grain-sys",
actor.WithShutdownTimeout(5*time.Second),
)
sysMod, err := NewActorSystemModule("test-grain-sys", map[string]any{"shutdownTimeout": "5s"})
if err != nil {
t.Fatalf("failed to create actor system: %v", err)
}
if err := sys.Start(ctx); err != nil {
if err := sysMod.Start(ctx); err != nil {
t.Fatalf("failed to start actor system: %v", err)
}
defer sys.Stop(ctx) //nolint:errcheck

factory := func(_ context.Context) (actor.Grain, error) {
return &BridgeGrain{
poolName: "test-pool",
handlers: handlers,
}, nil
defer sysMod.Stop(ctx) //nolint:errcheck
sys := sysMod.ActorSystem()
pool := &ActorPoolModule{
name: "test-pool",
handlers: handlers,
system: sysMod,
idleTimeout: 10 * time.Minute,
}

grainID, err := sys.GrainIdentity(ctx, "grain-1", factory,
actor.WithGrainDeactivateAfter(10*time.Minute),
)
grainID, err := pool.GetGrainIdentity(ctx, "grain-1")
if err != nil {
t.Fatalf("failed to get grain identity: %v", err)
}
Expand Down
30 changes: 30 additions & 0 deletions plugins/actors/grain_pool.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
package actors

import "sync"

const grainPoolRegistryID = "workflow-grain-pools"

// grainPoolRegistry keeps runtime services local to the actor system. GrainOf
// constructs zero-value grains, which resolve their pool during activation.
type grainPoolRegistry struct {
pools sync.Map
}

func (*grainPoolRegistry) ID() string { return grainPoolRegistryID }

// grainPoolReference carries only the pool name through grain recreation;
// registries, applications and loggers are resolved on the owning system.
type grainPoolReference struct {
poolName string
}

func (*grainPoolReference) ID() string { return "workflow-grain-pool" }

func (r *grainPoolReference) MarshalBinary() ([]byte, error) {
return []byte(r.poolName), nil
}

func (r *grainPoolReference) UnmarshalBinary(data []byte) error {
r.poolName = string(data)
return nil
}
14 changes: 3 additions & 11 deletions plugins/actors/module_pool.go
Original file line number Diff line number Diff line change
Expand Up @@ -244,17 +244,9 @@ func (m *ActorPoolModule) GetGrainIdentity(ctx context.Context, identity string)
return nil, fmt.Errorf("actor.pool %q: actor system not started", m.name)
}

factory := func(_ context.Context) (actor.Grain, error) {
return &BridgeGrain{
poolName: m.name,
handlers: m.handlers,
registry: m.stepRegistry,
app: m.app,
logger: m.logger,
}, nil
}

return m.system.ActorSystem().GrainIdentity(ctx, identity, factory,
m.system.grainPools.pools.Store(m.name, m)
return actor.GrainOf[*BridgeGrain](ctx, m.system.ActorSystem(), identity,
actor.WithGrainDependencies(&grainPoolReference{poolName: m.name}),
actor.WithGrainDeactivateAfter(m.idleTimeout),
)
}
Expand Down
6 changes: 6 additions & 0 deletions plugins/actors/module_system.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ type ActorSystemModule struct {
shutdownTimeout time.Duration
system actor.ActorSystem
logger *slog.Logger
grainPools grainPoolRegistry

// Default recovery policy
defaultSupervisor *supervisor.Supervisor
Expand Down Expand Up @@ -78,6 +79,7 @@ func (m *ActorSystemModule) Start(ctx context.Context) error {
opts := []actor.Option{
actor.WithShutdownTimeout(m.shutdownTimeout),
actor.WithDefaultSupervisor(m.defaultSupervisor),
actor.WithExtensions(&m.grainPools),
}

sys, err := actor.NewActorSystem(m.name, opts...)
Expand All @@ -88,6 +90,10 @@ func (m *ActorSystemModule) Start(ctx context.Context) error {
if err := sys.Start(ctx); err != nil {
return fmt.Errorf("actor.system %q: failed to start: %w", m.name, err)
}
if err := sys.Inject(&grainPoolReference{}); err != nil {
_ = sys.Stop(ctx)
return fmt.Errorf("actor.system %q: failed to register grain pool dependency: %w", m.name, err)
}

m.system = sys
return nil
Expand Down
Loading