Repository: Tochemey/goakt
Pull request #155 (2023-11-04 09:13:37)

refactor: add init timeout for actor
This will help enhance spawn actors

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/155?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Comparison is base [(`bf7be47`)](https://app.codecov.io/gh/Tochemey/goakt/commit/bf7be478bf9b88ff4cfd72c651a01d4d1ef9905c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 89.63% compared to head [(`9e6b448`)](https://app.codecov.io/gh/Tochemey/goakt/pull/155?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 89.68%.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #155      +/-   ##
==========================================
+ Coverage   89.63%   89.68%   +0.04%     
==========================================
  Files          41       41              
  Lines        3957     3973      +16     
==========================================
+ Hits         3547     3563      +16     
  Misses        328      328              
  Partials       82       82              
```


| [Files](https://app.codecov.io/gh/Tochemey/goakt/pull/155?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) | Coverage Δ | |
|---|---|---|
| [actors/actor\_system.go](https://app.codecov.io/gh/Tochemey/goakt/pull/155?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL2FjdG9yX3N5c3RlbS5nbw==) | `90.50% <100.00%> (+0.01%)` | :arrow_up: |
| [actors/option.go](https://app.codecov.io/gh/Tochemey/goakt/pull/155?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL29wdGlvbi5nbw==) | `100.00% <100.00%> (ø)` | |
| [actors/pid.go](https://app.codecov.io/gh/Tochemey/goakt/pull/155?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL3BpZC5nbw==) | `94.97% <100.00%> (+0.04%)` | :arrow_up: |
| [actors/pid\_option.go](https://app.codecov.io/gh/Tochemey/goakt/pull/155?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL3BpZF9vcHRpb24uZ28=) | `100.00% <100.00%> (ø)` | |


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/155?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).


Patch:
diff --git a/actors/actor_system.go b/actors/actor_system.go
index b6e68e7..1c92cd7 100644
--- a/actors/actor_system.go
+++ b/actors/actor_system.go
@@ -141,6 +141,9 @@ type actorSystem struct {
 	// Specifies the maximum of retries to attempt when the actor
 	// initialization fails. The default value is 5
 	actorInitMaxRetries int
+	// Specifies the actors initialization timeout
+	// The default value is 1s
+	actorInitTimeout time.Duration
 	// Specifies the supervisor strategy
 	supervisorStrategy StrategyDirective
 	// Specifies the telemetry config
@@ -355,6 +358,7 @@ func (x *actorSystem) Spawn(ctx context.Context, name string, actor Actor) (PID,
 		withMailbox(x.mailbox),
 		withStash(x.stashBuffer),
 		withEventsStream(x.eventsStream),
+		withInitTimeout(x.actorInitTimeout),
 		withTelemetry(x.telemetry))
 
 	// handle the error
diff --git a/actors/constants.go b/actors/constants.go
index d39435c..dd87c7b 100644
--- a/actors/constants.go
+++ b/actors/constants.go
@@ -51,6 +51,8 @@ const (
 	DefaultSupervisoryStrategy = StopDirective
 	// DefaultShutdownTimeout defines the default shutdown timeout
 	DefaultShutdownTimeout = 2 * time.Second
+	// DefaultInitTimeout defines the default init timeout
+	DefaultInitTimeout = time.Second
 
 	defaultMailboxSize = 4096
 	deadlettersTopic   = "topic.deadletters"
diff --git a/actors/option.go b/actors/option.go
index fc43b9f..32d8888 100644
--- a/actors/option.go
+++ b/actors/option.go
@@ -158,3 +158,10 @@ func WithPartitionHasher(hasher hash.Hasher) Option {
 		a.partitionHasher = hasher
 	})
 }
+
+// WithActorInitTimeout sets how long in seconds an actor start timeout
+func WithActorInitTimeout(timeout time.Duration) Option {
+	return OptionFunc(func(a *actorSystem) {
+		a.actorInitTimeout = timeout
+	})
+}
diff --git a/actors/pid.go b/actors/pid.go
index 8900a25..1be0f25 100644
--- a/actors/pid.go
+++ b/actors/pid.go
@@ -33,7 +33,7 @@ import (
 
 	"connectrpc.com/connect"
 	"connectrpc.com/otelconnect"
-	"github.com/cenkalti/backoff"
+	"github.com/flowchartsman/retry"
 	"github.com/pkg/errors"
 	"github.com/tochemey/goakt/eventstream"
 	internalpb "github.com/tochemey/goakt/internal/v1"
@@ -153,12 +153,16 @@ type pid struct {
 
 	// specifies how long the sender of a mail should wait to receive a reply
 	// when using SendReply. The default value is 5s
-	sendReplyTimeout atomic.Duration
+	replyTimeout atomic.Duration
 
 	// specifies the maximum of retries to attempt when the actor
-	// initialization fails. The default value is 5 seconds
+	// initialization fails. The default value is 5
 	initMaxRetries atomic.Int32
 
+	// specifies the init timeout. the default initialization timeout is
+	// 1s
+	initTimeout atomic.Duration
+
 	// shutdownTimeout specifies the graceful shutdown timeout
 	// the default value is 5 seconds
 	shutdownTimeout atomic.Duration
@@ -246,7 +250,8 @@ func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption
 	pid.stashCapacity.Store(0)
 	pid.isRunning.Store(false)
 	pid.passivateAfter.Store(DefaultPassivationTimeout)
-	pid.sendReplyTimeout.Store(DefaultReplyTimeout)
+	pid.replyTimeout.Store(DefaultReplyTimeout)
+	pid.initTimeout.Store(DefaultInitTimeout)
 
 	// set the custom options to override the default values
 	for _, opt := range opts {
@@ -457,7 +462,7 @@ func (p *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 		actor,
 		withInitMaxRetries(int(p.initMaxRetries.Load())),
 		withPassivationAfter(p.passivateAfter.Load()),
-		withSendReplyTimeout(p.sendReplyTimeout.Load()),
+		withSendReplyTimeout(p.replyTimeout.Load()),
 		withCustomLogger(p.logger),
 		withActorSystem(p.system),
 		withSupervisorStrategy(p.supervisorStrategy),
@@ -465,6 +470,7 @@ func (p *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 		withStash(p.stashCapacity.Load()),
 		withMailbox(p.mailbox.Clone()),
 		withEventsStream(p.eventsStream),
+		withInitTimeout(p.initTimeout.Load()),
 		withShutdownTimeout(p.shutdownTimeout.Load()))
 
 	// handle the error
@@ -516,7 +522,7 @@ func (p *pid) Ask(ctx context.Context, to PID, message proto.Message) (response
 	to.doReceive(context)
 
 	// await patiently to receive the response from the actor
-	for await := time.After(p.sendReplyTimeout.Load()); ; {
+	for await := time.After(p.replyTimeout.Load()); ; {
 		select {
 		case response = <-context.response:
 			return
@@ -775,12 +781,15 @@ func (p *pid) doReceive(ctx ReceiveContext) {
 func (p *pid) init(ctx context.Context) error {
 	// add some logging info
 	p.logger.Info("Initialization process has started...")
-	// create the exponential backoff object
-	expoBackoff := backoff.WithMaxRetries(backoff.NewExponentialBackOff(), uint64(p.initMaxRetries.Load()))
+	// create a context to handle the start timeout
+	cancelCtx, cancel := context.WithTimeout(ctx, p.initTimeout.Load())
+	// cancel the context when done
+	defer cancel()
+	// create a new retrier that will try a maximum of `initMaxRetries` times, with
+	// an initial delay of 100 ms and a maximum delay of 1 second
+	retrier := retry.NewRetrier(int(p.initMaxRetries.Load()), 100*time.Millisecond, time.Second)
 	// init the actor initialization receive
-	err := backoff.Retry(func() error {
-		return p.Actor.PreStart(ctx)
-	}, expoBackoff)
+	err := retrier.RunContext(cancelCtx, p.Actor.PreStart)
 	// handle backoff error
 	if err != nil {
 		return ErrInitFailure(err)
@@ -799,10 +808,11 @@ func (p *pid) reset() {
 	// reset the various settings of the PID
 	p.lastProcessingTime = atomic.Time{}
 	p.passivateAfter.Store(DefaultPassivationTimeout)
-	p.sendReplyTimeout.Store(DefaultReplyTimeout)
+	p.replyTimeout.Store(DefaultReplyTimeout)
 	p.shutdownTimeout.Store(DefaultShutdownTimeout)
 	p.initMaxRetries.Store(DefaultInitMaxRetries)
 	p.lastProcessingDuration.Store(0)
+	p.initTimeout.Store(0)
 	p.children = newPIDMap(10)
 	p.watchMen = slices.NewConcurrentSlice[*watchMan]()
 	p.telemetry = telemetry.New()
diff --git a/actors/pid_option.go b/actors/pid_option.go
index 77dcad6..31fd515 100644
--- a/actors/pid_option.go
+++ b/actors/pid_option.go
@@ -46,7 +46,7 @@ func withPassivationAfter(duration time.Duration) pidOption {
 // in a receive-reply pattern
 func withSendReplyTimeout(timeout time.Duration) pidOption {
 	return func(pid *pid) {
-		pid.sendReplyTimeout.Store(timeout)
+		pid.replyTimeout.Store(timeout)
 	}
 }
 
@@ -127,3 +127,10 @@ func withEventsStream(stream *eventstream.EventsStream) pidOption {
 		pid.eventsStream = stream
 	}
 }
+
+// withInitTimeout sets the init timeout
+func withInitTimeout(duration time.Duration) pidOption {
+	return func(pid *pid) {
+		pid.initTimeout.Store(duration)
+	}
+}
diff --git a/go.mod b/go.mod
index ba93b84..8f666fc 100644
--- a/go.mod
+++ b/go.mod
@@ -7,7 +7,6 @@ require (
 	connectrpc.com/otelconnect v0.6.0
 	github.com/buraksezer/olric v0.5.4
 	github.com/caarlos0/env/v10 v10.0.0
-	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/deckarep/golang-set/v2 v2.3.1
 	github.com/flowchartsman/retry v1.2.0
@@ -42,6 +41,7 @@ require (
 	github.com/armon/go-metrics v0.4.1 // indirect
 	github.com/bits-and-blooms/bitset v1.8.0 // indirect
 	github.com/buraksezer/consistent v0.10.0 // indirect
+	github.com/cenkalti/backoff v2.2.1+incompatible // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect


Test patch:
diff --git a/actors/option_test.go b/actors/option_test.go
index 1b7f39b..9757390 100644
--- a/actors/option_test.go
+++ b/actors/option_test.go
@@ -111,6 +111,11 @@ func TestOptions(t *testing.T) {
 			option:   WithPartitionHasher(hasher),
 			expected: actorSystem{partitionHasher: hasher},
 		},
+		{
+			name:     "WithActorInitTimeout",
+			option:   WithActorInitTimeout(2 * time.Second),
+			expected: actorSystem{actorInitTimeout: 2. * time.Second},
+		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
diff --git a/actors/pid_option_test.go b/actors/pid_option_test.go
index fb8e370..bc13169 100644
--- a/actors/pid_option_test.go
+++ b/actors/pid_option_test.go
@@ -61,7 +61,7 @@ func TestPIDOptions(t *testing.T) {
 		{
 			name:     "WithSendReplyTimeout",
 			option:   withSendReplyTimeout(time.Second),
-			expected: &pid{sendReplyTimeout: atomicDuration},
+			expected: &pid{replyTimeout: atomicDuration},
 		},
 		{
 			name:     "WithInitMaxRetries",
@@ -108,6 +108,11 @@ func TestPIDOptions(t *testing.T) {
 			option:   withEventsStream(eventsStream),
 			expected: &pid{eventsStream: eventsStream},
 		},
+		{
+			name:     "withInitTimeout",
+			option:   withInitTimeout(time.Second),
+			expected: &pid{initTimeout: atomicDuration},
+		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {