From 754231f55820b5d032eb5a3a93eb21ed8040c486 Mon Sep 17 00:00:00 2001 From: Asim Aslam Date: Mon, 21 Sep 2026 08:20:20 +0100 Subject: [PATCH] feat: wait for service dependencies with cancellation and backoff --- .../website/content/en/docs/client-server.md | 27 +++++ wait.go | 110 ++++++++++++++++++ wait_test.go | 107 +++++++++++++++++ 3 files changed, 244 insertions(+) create mode 100644 wait.go create mode 100644 wait_test.go diff --git a/internal/website/content/en/docs/client-server.md b/internal/website/content/en/docs/client-server.md index 0866d600c3..98815998f1 100644 --- a/internal/website/content/en/docs/client-server.md +++ b/internal/website/content/en/docs/client-server.md @@ -40,3 +40,30 @@ func main() { } } ``` + +## Wait for a dependency + +Startup order does not guarantee that a dependency has registered. Use the Go +helper with a caller deadline instead of maintaining a retry loop in each service: + +```go +ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) +defer cancel() +if err := micro.WaitForService(ctx, service, "identity", + micro.WaitBackoff(100*time.Millisecond, 2*time.Second), +); err != nil { + return err +} +``` + +The helper waits for a registered node with an address. Discovery does not prove +that the application is ready to answer requests. Add `micro.WaitProbe(func(ctx +context.Context) error { ... })` to perform a read-only health or initialization +RPC after discovery. The probe may run repeatedly and should honor its context. + +Registry errors, empty node lists, and probe failures are retried with capped +exponential backoff. Defaults are 100 milliseconds initially and 2 seconds at the +cap. Cancellation returns promptly even if a registry backend ignores context; +its one in-flight attempt may finish later. `errors.Is` can identify the context +cancellation/deadline and the last completed failure. This helper does not alter +service readiness endpoints or start a background dependency monitor. diff --git a/wait.go b/wait.go new file mode 100644 index 0000000000..09deef09ca --- /dev/null +++ b/wait.go @@ -0,0 +1,110 @@ +package micro + +import ( + "context" + "errors" + "fmt" + "time" + + "go-micro.dev/v6/registry" +) + +// WaitOptions controls dependency discovery and an optional readiness probe. +type WaitOptions struct { + InitialBackoff time.Duration + MaxBackoff time.Duration + Probe func(context.Context) error +} + +// WaitOption configures WaitForService. +type WaitOption func(*WaitOptions) + +// WaitBackoff sets the initial and maximum exponential retry delays. +func WaitBackoff(initial, maximum time.Duration) WaitOption { + return func(o *WaitOptions) { o.InitialBackoff = initial; o.MaxBackoff = maximum } +} + +// WaitProbe adds an application readiness check after discovery succeeds. +// The probe should be safe to repeat and honor its context, for example a read-only RPC. +func WaitProbe(probe func(context.Context) error) WaitOption { + return func(o *WaitOptions) { o.Probe = probe } +} + +// WaitForService waits for at least one registered node with an address, then +// runs the optional probe. Discovery alone does not guarantee RPC readiness. +// Retries use capped exponential backoff and stop when ctx is done. A registry +// backend or probe that ignores context may finish its in-flight attempt after +// this function returns; no additional attempts are started after cancellation. +func WaitForService(ctx context.Context, svc Service, name string, opts ...WaitOption) error { + if svc == nil || svc.Options().Registry == nil || name == "" { + return errors.New("wait for service: service, registry and dependency name are required") + } + options := WaitOptions{InitialBackoff: 100 * time.Millisecond, MaxBackoff: 2 * time.Second} + for _, opt := range opts { + opt(&options) + } + if options.InitialBackoff <= 0 || options.MaxBackoff < options.InitialBackoff { + return errors.New("wait for service: backoff must be positive and maximum must not be smaller than initial") + } + reg := svc.Options().Registry + delay := options.InitialBackoff + var lastErr error + for { + if err := ctx.Err(); err != nil { + return fmt.Errorf("wait for service %s: %w", name, errors.Join(err, lastErr)) + } + result := make(chan error, 1) + go func() { + if err := ctx.Err(); err != nil { + result <- err + return + } + services, err := reg.GetService(name, func(o *registry.GetOptions) { o.Context = ctx }) + if err == nil { + found := false + for _, service := range services { + if service == nil { + continue + } + for _, node := range service.Nodes { + if node != nil && node.Address != "" { + found = true + break + } + } + } + if !found { + err = registry.ErrNotFound + } else if options.Probe != nil { + if err = ctx.Err(); err == nil { + err = options.Probe(ctx) + } + } + } + result <- err + }() + select { + case <-ctx.Done(): + return fmt.Errorf("wait for service %s: %w", name, errors.Join(ctx.Err(), lastErr)) + case lastErr = <-result: + if err := ctx.Err(); err != nil { + return fmt.Errorf("wait for service %s: %w", name, errors.Join(err, lastErr)) + } + if lastErr == nil { + return nil + } + } + timer := time.NewTimer(delay) + select { + case <-ctx.Done(): + timer.Stop() + return fmt.Errorf("wait for service %s: %w", name, errors.Join(ctx.Err(), lastErr)) + case <-timer.C: + } + if delay > options.MaxBackoff/2 { + delay = options.MaxBackoff + } else { + delay *= 2 + } + } +} diff --git a/wait_test.go b/wait_test.go new file mode 100644 index 0000000000..4ee623f420 --- /dev/null +++ b/wait_test.go @@ -0,0 +1,107 @@ +package micro + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "go-micro.dev/v6/registry" +) + +type waitingRegistry struct { + registry.Registry + get func(string, ...registry.GetOption) ([]*registry.Service, error) +} + +func (r waitingRegistry) GetService(name string, opts ...registry.GetOption) ([]*registry.Service, error) { + return r.get(name, opts...) +} + +func TestWaitForServiceRetriesDiscoveryAndProbe(t *testing.T) { + var lookups, probes int + transient := errors.New("not ready") + reg := waitingRegistry{get: func(_ string, opts ...registry.GetOption) ([]*registry.Service, error) { + lookups++ + var options registry.GetOptions + for _, opt := range opts { + opt(&options) + } + if options.Context == nil { + t.Error("missing lookup context") + } + if lookups == 1 { + return nil, transient + } + if lookups == 2 { + return []*registry.Service{{Nodes: []*registry.Node{{Address: ""}}}}, nil + } + return []*registry.Service{{Nodes: []*registry.Node{{Address: "localhost:1"}}}}, nil + }} + svc := NewService("caller", Registry(reg)) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + err := WaitForService(ctx, svc, "dependency", WaitBackoff(time.Millisecond, 2*time.Millisecond), WaitProbe(func(context.Context) error { + probes++ + if probes == 1 { + return transient + } + return nil + })) + if err != nil || lookups != 4 || probes != 2 { + t.Fatalf("err=%v lookups=%d probes=%d", err, lookups, probes) + } +} + +func TestWaitForServiceCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + entered, release := make(chan struct{}), make(chan struct{}) + defer close(release) + var calls atomic.Int32 + reg := waitingRegistry{get: func(string, ...registry.GetOption) ([]*registry.Service, error) { + calls.Add(1) + close(entered) + <-release + return nil, registry.ErrNotFound + }} + svc := NewService("caller", Registry(reg)) + done := make(chan error, 1) + go func() { done <- WaitForService(ctx, svc, "dependency") }() + <-entered + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("cancellation blocked behind registry") + } + if calls.Load() != 1 { + t.Fatal("unexpected extra lookup") + } +} + +func TestWaitForServiceCanceledBeforeLookup(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + reg := waitingRegistry{get: func(string, ...registry.GetOption) ([]*registry.Service, error) { + t.Error("lookup after cancellation") + return nil, nil + }} + if err := WaitForService(ctx, NewService("caller", Registry(reg)), "dependency"); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } +} + +func TestWaitForServiceDeadlinePreservesDiscoveryError(t *testing.T) { + cause := errors.New("registry unavailable") + reg := waitingRegistry{get: func(string, ...registry.GetOption) ([]*registry.Service, error) { return nil, cause }} + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + err := WaitForService(ctx, NewService("caller", Registry(reg)), "dependency", WaitBackoff(time.Millisecond, time.Millisecond)) + if !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, cause) { + t.Fatalf("lost error cause: %v", err) + } +}