feat: wait for service dependencies with cancellation and backoff

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.

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
+		}
+	}
+}

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)
+	}
+}