Merge pull request 'Ask the plans again when the moved path starts before its runtime answers (issue 943)' (#336) from fix/943-moved-path-waits-for-runtime into main

2026-10-16 16:55:41 +00:00
parent 832a0afd18 6005c6001e
commit 3817099b2c
4 changed files with 174 additions and 4 deletions
@@ -0,0 +1,62 @@
+package main
+
+// The moved path started before its runtime answers asks again until it does (issue 943).
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/provider"
+)
+
+// earlyBus refuses its first watches as the runtime does before Serve runs, then answers as the bucket bus does.
+type earlyBus struct {
+	*bucketBus
+	refuse int
+}
+
+func (b *earlyBus) Watch(ctx context.Context, bucket string, handle func(string, []byte)) error {
+	if b.refuse > 0 {
+		b.refuse--
+		return errors.New("before Serve: a bundle reaches the bus through the runtime that launched it")
+	}
+	return b.bucketBus.Watch(ctx, bucket, handle)
+}
+
+func TestAMovedPathStartedTooEarlyAsksAgainUntilTheRuntimeAnswers(t *testing.T) {
+	w := newMovedWorld(t, map[[2]string]string{})
+	w.run.FirstRetry = time.Millisecond
+	w.run.loaded = false
+	bus := &earlyBus{bucketBus: w.bus, refuse: 3}
+	p := provider.New(provider.Options{Node: "novox", Module: "keycloak", Bus: w.bus,
+		Vault: provider.ConsumersFile{Paths: []string{w.file}}, Make: w.run.Make, Gone: w.run.Gone,
+		Every: time.Hour, Logf: func(string, ...any) {}})
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- w.run.Run(ctx, p, bus) }()
+	deadline := time.After(time.Second)
+	for {
+		w.run.mu.Lock()
+		loaded := w.run.loaded
+		w.run.mu.Unlock()
+		if loaded {
+			break
+		}
+		select {
+		case err := <-done:
+			cancel()
+			t.Fatalf("Run stopped before the runtime answered: %v", err)
+		case <-deadline:
+			cancel()
+			t.Fatal("the plans were never watched")
+		case <-time.After(time.Millisecond):
+		}
+	}
+	cancel()
+	<-done
+	if bus.refuse != 0 {
+		t.Fatalf("%d refusals left", bus.refuse)
+	}
+}
@@ -52,6 +52,9 @@ func valuesOf(e provider.Entry) map[string]any {
 type movedRun struct {
 	h *Harness

+	// FirstRetry is the first wait before a watch asked too early is asked again (issue 943); a second when zero.
+	FirstRetry time.Duration
+
 	mu          sync.Mutex // guards assignments and loaded
 	assignments map[string][]byte
 	loaded      bool
@@ -231,9 +234,29 @@ func (m *movedRun) watch(ctx context.Context, bus provider.Bus) error {
 }

 // Run answers the consumers through the helper and retires through the harness's rules, until ctx ends.
+//
+// The bus is reached through the runtime that launched this bundle, which answers only once Serve runs, and Run starts
+// beside Serve: a watch asked too early is asked again, after a wait that doubles from FirstRetry up to a minute,
+// until it is answered or ctx ends, so a start that loses the race never leaves the consumers unanswered (issue 943).
 func (m *movedRun) Run(ctx context.Context, p *provider.Provider, bus provider.Bus) error {
-	if err := m.watch(ctx, bus); err != nil {
-		return err
+	wait := m.FirstRetry
+	if wait <= 0 {
+		wait = time.Second
+	}
+	for tries := 1; ; tries++ {
+		err := m.watch(ctx, bus)
+		if err == nil {
+			break
+		}
+		if tries == 4 || tries%20 == 0 {
+			m.h.say("the plans cannot be watched yet (%d tries), asking again: %v", tries, err)
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(wait):
+		}
+		wait = min(wait*2, time.Minute)
 	}
 	go func() {
 		for {
@@ -0,0 +1,62 @@
+package main
+
+// The moved path started before its runtime answers asks again until it does (issue 943).
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/provider"
+)
+
+// earlyBus refuses its first watches as the runtime does before Serve runs, then answers as the bucket bus does.
+type earlyBus struct {
+	*bucketBus
+	refuse int
+}
+
+func (b *earlyBus) Watch(ctx context.Context, bucket string, handle func(string, []byte)) error {
+	if b.refuse > 0 {
+		b.refuse--
+		return errors.New("before Serve: a bundle reaches the bus through the runtime that launched it")
+	}
+	return b.bucketBus.Watch(ctx, bucket, handle)
+}
+
+func TestAMovedPathStartedTooEarlyAsksAgainUntilTheRuntimeAnswers(t *testing.T) {
+	w := newMovedWorld(t, map[[2]string]string{})
+	w.run.FirstRetry = time.Millisecond
+	w.run.loaded = false
+	bus := &earlyBus{bucketBus: w.bus, refuse: 3}
+	p := provider.New(provider.Options{Node: "novox", Module: "postgres", Bus: w.bus,
+		Vault: provider.ConsumersFile{Paths: []string{w.file}}, Make: w.run.Make, Gone: w.run.Gone,
+		Every: time.Hour, Logf: func(string, ...any) {}})
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- w.run.Run(ctx, p, bus) }()
+	deadline := time.After(time.Second)
+	for {
+		w.run.mu.Lock()
+		loaded := w.run.loaded
+		w.run.mu.Unlock()
+		if loaded {
+			break
+		}
+		select {
+		case err := <-done:
+			cancel()
+			t.Fatalf("Run stopped before the runtime answered: %v", err)
+		case <-deadline:
+			cancel()
+			t.Fatal("the plans were never watched")
+		case <-time.After(time.Millisecond):
+		}
+	}
+	cancel()
+	<-done
+	if bus.refuse != 0 {
+		t.Fatalf("%d refusals left", bus.refuse)
+	}
+}
@@ -38,6 +38,9 @@ const servedPort = 5432
 type movedRun struct {
 	h *Harness

+	// FirstRetry is the first wait before a watch asked too early is asked again (issue 943); a second when zero.
+	FirstRetry time.Duration
+
 	mu          sync.Mutex // guards assignments and loaded
 	assignments map[string][]byte
 	loaded      bool
@@ -252,9 +255,29 @@ func (m *movedRun) watch(ctx context.Context, bus provider.Bus) error {
 }

 // Run answers the consumers through the helper and retires through the harness's rules, until ctx ends.
+//
+// The bus is reached through the runtime that launched this bundle, which answers only once Serve runs, and Run starts
+// beside Serve: a watch asked too early is asked again, after a wait that doubles from FirstRetry up to a minute,
+// until it is answered or ctx ends, so a start that loses the race never leaves the consumers unanswered (issue 943).
 func (m *movedRun) Run(ctx context.Context, p *provider.Provider, bus provider.Bus) error {
-	if err := m.watch(ctx, bus); err != nil {
-		return err
+	wait := m.FirstRetry
+	if wait <= 0 {
+		wait = time.Second
+	}
+	for tries := 1; ; tries++ {
+		err := m.watch(ctx, bus)
+		if err == nil {
+			break
+		}
+		if tries == 4 || tries%20 == 0 {
+			m.h.say("the plans cannot be watched yet (%d tries), asking again: %v", tries, err)
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(wait):
+		}
+		wait = min(wait*2, time.Minute)
 	}
 	go func() {
 		for {