diff --git a/example/go.mod b/example/go.mod index 445d47fd..0f630214 100644 --- a/example/go.mod +++ b/example/go.mod @@ -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 diff --git a/example/go.sum b/example/go.sum index 59e0ba3e..0bee561f 100644 --- a/example/go.sum +++ b/example/go.sum @@ -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= diff --git a/go.mod b/go.mod index 3285f078..f13ca0d4 100644 --- a/go.mod +++ b/go.mod @@ -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 @@ -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 diff --git a/go.sum b/go.sum index c0a6460d..a1efb768 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/plugins/actors/bridge_actor.go b/plugins/actors/bridge_actor.go index ac7ab6de..1bf6cbf0 100644 --- a/plugins/actors/bridge_actor.go +++ b/plugins/actors/bridge_actor.go @@ -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) } diff --git a/plugins/actors/bridge_actor_test.go b/plugins/actors/bridge_actor_test.go index efffbfe0..2908a3c8 100644 --- a/plugins/actors/bridge_actor_test.go +++ b/plugins/actors/bridge_actor_test.go @@ -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() @@ -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) } diff --git a/plugins/actors/grain_pool.go b/plugins/actors/grain_pool.go new file mode 100644 index 00000000..71e4d5bf --- /dev/null +++ b/plugins/actors/grain_pool.go @@ -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 +} diff --git a/plugins/actors/module_pool.go b/plugins/actors/module_pool.go index dc8d6214..970786c7 100644 --- a/plugins/actors/module_pool.go +++ b/plugins/actors/module_pool.go @@ -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), ) } diff --git a/plugins/actors/module_system.go b/plugins/actors/module_system.go index a314294c..7859c79d 100644 --- a/plugins/actors/module_system.go +++ b/plugins/actors/module_system.go @@ -18,6 +18,7 @@ type ActorSystemModule struct { shutdownTimeout time.Duration system actor.ActorSystem logger *slog.Logger + grainPools grainPoolRegistry // Default recovery policy defaultSupervisor *supervisor.Supervisor @@ -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...) @@ -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