diff --git a/modules/keycloak/cmd/keycloak-provider/early_test.go b/modules/keycloak/cmd/keycloak-provider/early_test.go new file mode 100644 index 00000000..9a79a729 --- /dev/null +++ b/modules/keycloak/cmd/keycloak-provider/early_test.go @@ -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) + } +} diff --git a/modules/keycloak/cmd/keycloak-provider/moved.go b/modules/keycloak/cmd/keycloak-provider/moved.go index 34ce390f..7e63eba5 100644 --- a/modules/keycloak/cmd/keycloak-provider/moved.go +++ b/modules/keycloak/cmd/keycloak-provider/moved.go @@ -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 { diff --git a/modules/postgres/cmd/postgres-provider/early_test.go b/modules/postgres/cmd/postgres-provider/early_test.go new file mode 100644 index 00000000..592c8b53 --- /dev/null +++ b/modules/postgres/cmd/postgres-provider/early_test.go @@ -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) + } +} diff --git a/modules/postgres/cmd/postgres-provider/moved.go b/modules/postgres/cmd/postgres-provider/moved.go index a6ec5b79..25fd1ea2 100644 --- a/modules/postgres/cmd/postgres-provider/moved.go +++ b/modules/postgres/cmd/postgres-provider/moved.go @@ -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 {