feat: add support for notification-based WaitForConfig

[meling]

We currently have a test helper in testing_shared.go: WaitForConfigCondition(), which uses a timer-delay loop to repeatedly check whether all servers have joined the configuration. This is useful in tests to ensure that we know when we have a full configuration. But it can also be useful for users who want to wait until a full configuration is ready before accepting quorum calls.

However, since inboundManager.rebuildConfig already knows exactly when the configuration changes and holds the lock at that point. Rather than forcing callers to poll Config() in a loop, the inboundManager should broadcast each configuration change to waiting goroutines the moment it is rebuilt. This eliminates unnecessary latency and CPU overhead.

[meling]

Below is the third iteration of the design doc. This will not be implemented because it is too invasive and adds too much noise for marginal benefit. I'm adding it here for posterity.

Problem

inboundManager.rebuildConfig already knows exactly when the configuration changes and holds the lock at that point.
Rather than forcing callers to poll Config() in a loop, the inboundManager should push each new configuration to registered waiters the moment it is rebuilt.
This eliminates unnecessary latency and CPU overhead.
This design doc describes how to complete that idea correctly.

---

Goals

• Replace polling with push-based notifications from rebuildConfig.
• Support multiple concurrent waiters without any waiter blocking others.
• Cover both Config (known peer servers) and ClientConfig (anonymous back-channel clients).
• Keep WaitForConfig as a standalone function that works with any channel source.
• Avoid goroutine or channel leaks.
• Retain the existing Config() and ClientConfig() snapshot methods unchanged.

---

Proposed design

Core mechanism: per-subscriber buffered channels

Each caller that wants push notifications gets its own chan Configuration with a buffer of 1.
rebuildConfig does a non-blocking send to every registered subscriber after updating im.config or im.clientConfig.
If a subscriber's buffer is full (the previous update has not been consumed yet), the send is skipped.
The subscriber will receive the next update when its buffer empties; it therefore converges to the latest configuration even if it missed intermediate states.
This is safe because WaitForConfig only needs to know that something changed — it re-evaluates the condition after each notification.

rebuildConfig()
  ├── update im.config / im.clientConfig
  └── for each ch in im.configSubs / im.clientConfigSubs:
          select { case ch <- cfg: default: }

Subscriber registration and cleanup

The subscribe function accepts a context.Context.
On call it:

1. Acquires the write lock.
2. Creates a chan Configuration with buffer 1.
3. Sends the current snapshot so the caller always has an initial value.
4. Appends the channel to the appropriate subscriber slice.
5. Releases the lock.
6. Spawns a cleanup goroutine that waits for ctx.Done(), then acquires the write lock, removes the channel from the slice, and closes it.

Closing the channel is the signal to WaitForConfig that the subscription ended (either because the system stopped or the caller's context was cancelled).

Changes to inboundManager

Add two subscriber slices to the struct:

type inboundManager struct {
    // ... existing fields unchanged ...
    configSubs      []chan Configuration // registered Config subscribers; protected by mu
    clientConfigSubs []chan Configuration // registered ClientConfig subscribers; protected by mu
}

Add an unexported subscribe helper:

// subscribeConfig returns a channel that immediately receives the current known-peer
// Configuration and then receives each subsequent update until ctx is cancelled.
// The channel is closed when ctx is done or when the inboundManager is closed.
func (im *inboundManager) subscribeConfig(ctx context.Context) <-chan Configuration {
    ch := make(chan Configuration, 1)
    im.mu.Lock()
    ch <- im.config
    im.configSubs = append(im.configSubs, ch)
    im.mu.Unlock()
    go func() {
        <-ctx.Done()
        im.mu.Lock()
        im.configSubs = removeSubscriber(im.configSubs, ch)
        im.mu.Unlock()
        close(ch)
    }()
    return ch
}

// subscribeClientConfig is the ClientConfig equivalent.
func (im *inboundManager) subscribeClientConfig(ctx context.Context) <-chan Configuration { ... }

Add a helper to remove a subscriber from a slice without preserving order (swap with last, re-slice):

func removeSubscriber(subs []chan Configuration, ch chan Configuration) []chan Configuration {
    for i, s := range subs {
        if s == ch {
            subs[i] = subs[len(subs)-1]
            return subs[:len(subs)-1]
        }
    }
    return subs
}

Update rebuildConfig to fan out after updating the slices (still under the existing lock):

func (im *inboundManager) rebuildConfig() {
    // ... existing rebuild logic unchanged ...

    // Fan out to subscribers (non-blocking; slow consumers get next update).
    notify(im.configSubs, im.config)
    notify(im.clientConfigSubs, im.clientConfig)
}

func notify(subs []chan Configuration, cfg Configuration) {
    for _, ch := range subs {
        select {
        case ch <- cfg:
        default:
        }
    }
}

On shutdown, close all subscriber channels so that WaitForConfig returns.
The existing inboundManager.Close (or the Server.Stop path) should drain and close any remaining subscribers:

func (im *inboundManager) closeSubscribers() {
    im.mu.Lock()
    defer im.mu.Unlock()
    for _, ch := range im.configSubs {
        close(ch)
    }
    im.configSubs = nil
    for _, ch := range im.clientConfigSubs {
        close(ch)
    }
    im.clientConfigSubs = nil
}

New WaitForConfig signature

Replace the current config func() Configuration parameter with changes <-chan Configuration:

// WaitForConfig blocks until cond returns true for a received Configuration,
// or until ctx is cancelled or changes is closed.
// The changes channel is obtained from [System.ConfigChanges] or [System.ClientConfigChanges].
// The initial Configuration is always the first value received on the channel,
// so cond may already be satisfied before any peer connects.
func WaitForConfig(ctx context.Context, changes <-chan Configuration, cond func(Configuration) bool) error {
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case cfg, ok := <-changes:
            if !ok {
                return ctx.Err() // channel closed: subscriber context done or system stopped
            }
            if cond(cfg) {
                return nil
            }
        }
    }
}

Expose subscription at System level

System gains two methods that delegate to the inboundManager:

// ConfigChanges returns a channel that immediately receives the current known-peer
// Configuration and then receives each subsequent update.
// The channel is closed when ctx is cancelled.
func (s *System) ConfigChanges(ctx context.Context) <-chan Configuration {
    return s.srv.subscribeConfig(ctx)
}

// ClientConfigChanges returns a channel that immediately receives the current
// client-peer Configuration and then receives each subsequent update.
// The channel is closed when ctx is cancelled.
func (s *System) ClientConfigChanges(ctx context.Context) <-chan Configuration {
    return s.srv.subscribeClientConfig(ctx)
}

Callers then look like:

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := gorums.WaitForConfig(ctx, sys.ConfigChanges(ctx), func(cfg gorums.Configuration) bool {
    return cfg.Size() == len(peers)
}); err != nil {
    log.Fatal("peers did not connect in time:", err)
}

Note that passing the same ctx to both ConfigChanges and WaitForConfig is intentional: cancellation is handled by whichever fires first — the select on ctx.Done() in WaitForConfig or the closed channel from the cleanup goroutine.

---

Design questions and decisions

Why per-subscriber channels rather than a shared broadcast channel?

A single shared channel means only one goroutine receives each update.
Per-subscriber channels allow any number of concurrent WaitForConfig calls — a common pattern when, for example, a test helper and application startup code both wait for the cluster to form.

Why buffer size 1?

Buffer size 1 guarantees that a non-blocking send in rebuildConfig never stalls and never drops the current value while the subscriber is processing the previous update.
It is sufficient because WaitForConfig only needs to know that a change happened, not every intermediate state.

Why a context parameter on the subscribe method rather than a stop function?

A context composes naturally with context.WithTimeout, context.WithCancel, and context.WithDeadline.
It avoids the need for the caller to track a separate cancel function.
The cleanup goroutine is lightweight and exits immediately when the context is cancelled.

Closing the channel on system shutdown

When Server.Stop is called before all WaitForConfig calls complete, calling closeSubscribers ensures those callers unblock.
WaitForConfig detects a closed channel via the ok check and returns ctx.Err() (which is nil if the caller did not cancel the context, or an error if it timed out).
If a more precise "system stopped" error is wanted, a dedicated sentinel error can be returned — but this can be deferred.

What happens to onConfigChange?

onConfigChange is left unchanged.
It serves a different purpose (e.g. triggering Raft leader elections or similar callbacks that must run synchronously after each config change) and is invoked while the lock is held with the explicit caution that it must not call back into Config.
The subscriber channels are populated independently.

Naming

Config2 / ClientConfig2 were placeholders proposed during design.
The final names are ConfigChanges / ClientConfigChanges on System, matching Go conventions for channel-returning methods that emit a stream of values (cf. time.Tick, fsnotify.Watcher.Events).
The internal inboundManager methods are subscribeConfig / subscribeClientConfig (unexported, verb-first).

---

What changes and what stays the same

Item	Change
inboundManager.Config()	Unchanged
inboundManager.ClientConfig()	Unchanged
inboundManager.configSubs	New field
inboundManager.clientConfigSubs	New field
inboundManager.subscribeConfig(ctx)	New method
inboundManager.subscribeClientConfig(ctx)	New method
inboundManager.rebuildConfig()	Extended to fan out to subscribers
inboundManager.closeSubscribers()	New method; called from Server.Stop
System.ConfigChanges(ctx)	New method
System.ClientConfigChanges(ctx)	New method
WaitForConfigCondition (test helper)	Unchanged (uses Config() snapshot; no need to change)

---

Implementation steps

1. Add configSubs and clientConfigSubs fields to inboundManager.
2. Implement subscribeConfig(ctx) and subscribeClientConfig(ctx) on inboundManager.
3. Implement the notify helper function and removeSubscriber helper.
4. Update rebuildConfig to call notify for both subscriber slices.
5. Implement closeSubscribers and wire it into the Server.Stop path.
6. Implement WaitForConfig in config.go to accept <-chan Configuration.
7. Add ConfigChanges and ClientConfigChanges to System.
8. Update the doc comment in config.go for WaitForConfig.
9. Update existing call sites (tests, examples) to pass sys.ConfigChanges(ctx) instead of sys.Config.
10. Write tests for: condition already met on entry; condition met after one notification; context cancelled before condition; system stopped before condition; multiple concurrent waiters.
11. Update doc/user-guide.md with a WaitForConfig example covering both ConfigChanges and ClientConfigChanges.

[meling]

Here is the fifth iteration of the design doc. This avoids tracking subscribers manually. It broadcasts using the channel close primitive as the recommended approach instead of sync.Cond.Broadcast(). I plan to implement this variant.

Problem

inboundManager.rebuildConfig already knows exactly when the configuration changes and holds the lock at that point. Rather than forcing callers to poll Config() in a loop, the inboundManager should broadcast each configuration change to waiting goroutines the moment it is rebuilt. This eliminates unnecessary latency and CPU overhead.

---

Goals

• Replace polling with push-based notifications from rebuildConfig.
• Support multiple concurrent waiters without any waiter blocking others.
• Cover both Config (known peer servers) and ClientConfig (anonymous back-channel clients).
• Avoid goroutine or channel leaks.
• Keep the API surface minimal: one method per config type on System, one context.Context parameter.
• Retain the existing Config() and ClientConfig() snapshot methods unchanged.

---

Proposed design

Core mechanism: close-and-replace broadcast channel

The inboundManager holds a single chan struct{} that acts as a broadcast signal.
When rebuildConfig updates the configuration, it closes the current channel (waking all waiters) and replaces it with a fresh one for the next change.

The Go documentation for sync.Cond itself recommends this pattern over condition variables:
"For many simple use cases, users will be better off using channels than a Cond (Broadcast corresponds to closing a channel)."

rebuildConfig()
  ├── update im.config / im.clientConfig
  ├── close(im.configCh)                    // broadcast: wake all waiters
  └── im.configCh = make(chan struct{})      // replace for next broadcast

Waiters grab the channel reference under the read lock, release the lock, then select on the channel and ctx.Done().
On wake, they re-acquire the read lock, re-check their condition, and either return or wait on the new channel.

This pattern is safe because:

• Closing a channel wakes all goroutines blocked on receive — true broadcast with no subscriber tracking.
• A receive on an already-closed channel returns immediately — no race between close and select.
• Each waiter holds its own reference to the channel that was current when it last checked — multiple close-and-replace cycles between checks are handled correctly because the waiter's captured channel will have been closed.

Changes to inboundManager

Add two fields to the struct:

type inboundManager struct {
    // ... existing fields unchanged ...
    configCh chan struct{} // closed and replaced on each config/clientConfig change; protected by mu
    stopped  bool         // set on shutdown to unblock waiters; protected by mu
}

Initialize the channel in the constructor:

func newInboundManager(myID uint32, opt NodeListOption, sendBuffer uint, onConfigChange func(Configuration), handler stream.RequestHandler) *inboundManager {
    im := &inboundManager{
        // ... existing field initialization unchanged ...
        configCh: make(chan struct{}),
    }
    // ... rest unchanged ...
}

Add the broadcast to rebuildConfig (still under the existing write lock):

func (im *inboundManager) rebuildConfig() {
    // ... existing rebuild logic unchanged ...

    // Broadcast config change to all waiters.
    // Guard: after close() has been called, configCh is already closed
    // and must not be closed again. rebuildConfig may still be called
    // during shutdown as peers disconnect.
    if !im.stopped {
        close(im.configCh)
        im.configCh = make(chan struct{})
    }
}

Add the waitForConfig method:

// waitForConfig blocks until cond returns true for the configuration returned
// by getConfig, or until ctx is cancelled or the inboundManager is closed.
func (im *inboundManager) waitForConfig(ctx context.Context, getConfig func() Configuration, cond func(Configuration) bool) error {
    im.mu.RLock()
    for {
        if cond(getConfig()) {
            im.mu.RUnlock()
            return nil
        }
        if im.stopped {
            im.mu.RUnlock()
            return ErrStopped
        }
        ch := im.configCh
        im.mu.RUnlock()
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-ch:
        }
        im.mu.RLock()
    }
}

Add a shutdown method that unblocks all waiters:

// close signals all waiters to stop and prevents new waits from blocking.
// Called from System.Stop.
func (im *inboundManager) close() {
    im.mu.Lock()
    defer im.mu.Unlock()
    if !im.stopped {
        im.stopped = true
        close(im.configCh)
    }
}

Add a sentinel error for the stopped case:

// ErrStopped is returned by WaitForConfig and WaitForClientConfig when the
// system is stopped before the condition is met.
var ErrStopped = errors.New("gorums: system stopped")

New methods on System

System gains two methods that delegate to the inboundManager:

// WaitForConfig blocks until cond returns true for the current known-peer
// [Configuration], or until ctx is cancelled or the system is stopped.
// The condition is checked immediately against the current configuration,
// so it may return without blocking if the condition is already satisfied.
func (s *System) WaitForConfig(ctx context.Context, cond func(Configuration) bool) error {
    return s.srv.waitForConfig(ctx, s.srv.Config, cond)
}

// WaitForClientConfig blocks until cond returns true for the current
// client-peer [Configuration], or until ctx is cancelled or the system is stopped.
// The condition is checked immediately against the current configuration,
// so it may return without blocking if the condition is already satisfied.
func (s *System) WaitForClientConfig(ctx context.Context, cond func(Configuration) bool) error {
    return s.srv.waitForConfig(ctx, s.srv.ClientConfig, cond)
}

Wire the close method into System.Stop:

func (s *System) Stop() (errs error) {
    s.srv.close() // unblock any WaitForConfig callers
    s.srv.Stop()
    _ = s.lis.Close()
    for _, closer := range s.closers {
        errs = errors.Join(errs, closer.Close())
    }
    return errs
}

Caller example

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := sys.WaitForConfig(ctx, func(cfg gorums.Configuration) bool {
    return cfg.Size() == len(peers)
}); err != nil {
    log.Fatal("peers did not connect in time:", err)
}

---

Design decisions

Why close-and-replace rather than per-subscriber channels?

Per-subscriber channels require tracking each subscriber, cleaning up on cancellation, and fanning out on each change.
Close-and-replace uses a single channel that broadcasts to all waiters simultaneously via close().
No subscriber list, no cleanup goroutines, no fan-out loop.

Why not sync.Cond?

sync.Cond cannot be used with select, making it impossible to compose with context.Context cancellation without spawning a helper goroutine per waiter.
The Go documentation itself discourages sync.Cond in favor of channels for simple use cases.

Why a single channel for both Config and ClientConfig?

Using one channel means a WaitForClientConfig waiter gets woken on Config changes (and vice versa).
This is a spurious wakeup — the waiter re-checks its condition and goes back to waiting.
The cost is negligible: config changes are infrequent (peer connect/disconnect events), and the condition check is a simple function call.
Using separate channels would double the channel management for no practical benefit.

Why methods on System rather than a standalone function?

The close-and-replace channel is internal to inboundManager.
A standalone function would need to accept *System and delegate to it anyway, making it a trivial wrapper.
Methods on System are the natural Go API — the receiver holds the state, and the caller needs only one ctx parameter.

Why a stopped field?

When System.Stop() is called, peer disconnections trigger rebuildConfig which broadcasts.
But if a waiter's condition cannot be met after shutdown (e.g., waiting for 3 peers when all have disconnected), it would wait indefinitely if the caller used context.Background() with no timeout.

The stopped flag ensures waitForConfig returns ErrStopped promptly after shutdown, regardless of the caller's context.
The close() method is idempotent — calling it multiple times is safe.

What happens to onConfigChange?

onConfigChange is left unchanged.
It serves a different purpose (e.g., triggering Raft leader elections or similar callbacks that must run synchronously after each config change) and is invoked while the lock is held.
The broadcast channel is independent.

What about WaitForConfigCondition (test helper)?

WaitForConfigCondition in testing_shared.go uses polling and works fine for tests.
It remains unchanged.
Tests may optionally migrate to the new methods, but this is not required.

---

Correctness analysis

1. Condition already met on first check.
   Returns immediately without entering select. No channel interaction.

2. Waiter checks condition (not met), grabs ch, releases RLock, enters select.
   Normal case. Waiter blocks until ch is closed by rebuildConfig.

3. rebuildConfig closes ch between condition check and select.
   Selecting on a closed channel returns immediately. Waiter re-checks condition.

4. Multiple rebuildConfig calls between condition check and select.
   The waiter's captured ch was closed in the first call. Select returns immediately.
   On re-check, the waiter sees the latest config and grabs the latest ch.

5. Context cancelled while waiting.
   The select fires on ctx.Done(). Returns ctx.Err().

6. System stopped while waiting.
   close() sets stopped = true and closes configCh. Waiter wakes, re-checks, sees stopped, returns ErrStopped.

7. Multiple concurrent waiters.
   Each holds its own reference to the same ch. close(ch) wakes all of them simultaneously. Each independently re-checks its own condition.

---

What changes and what stays the same

Item	Change
inboundManager.Config()	Unchanged
inboundManager.ClientConfig()	Unchanged
inboundManager.configCh	New field
inboundManager.stopped	New field
inboundManager.rebuildConfig()	Extended: close and replace configCh (2 lines)
inboundManager.waitForConfig()	New method
inboundManager.close()	New method; called from System.Stop
System.WaitForConfig(ctx, cond)	New method
System.WaitForClientConfig(ctx, cond)	New method
System.Stop()	Extended: call close() before stopping server
ErrStopped	New sentinel error
onConfigChange	Unchanged
WaitForConfigCondition (test helper)	Unchanged

---

Implementation steps

1. Add configCh and stopped fields to inboundManager.
2. Initialize configCh in newInboundManager.
3. Add close-and-replace to rebuildConfig (2 lines).
4. Implement waitForConfig on inboundManager.
5. Implement close on inboundManager.
6. Add ErrStopped sentinel error.
7. Add WaitForConfig and WaitForClientConfig to System.
8. Wire close() into System.Stop.
9. Update existing call sites (tests, examples) where appropriate.
10. Write tests: condition already met on entry; condition met after one notification; context cancelled before condition; system stopped before condition; multiple concurrent waiters.
11. Update doc/user-guide.md with a WaitForConfig example covering both WaitForConfig and WaitForClientConfig.