issue 974: a rollout advances on events (a look asked by each plan saved, report and expired silent key), the 30 s ticker goes, and each tier send is said as sent

cmd/mesh-controller/rollout_silent_bus.go

@@ -49,8 +49,10 @@ func parseSilentKey(key string) (silentWait, bool) {
 
 // silentClock is the bus's side of the wait for a machine's report.
 type silentClock interface {
-	// Arm puts a wait's key, to live ttl: made again if the machine is sent again.
-	Arm(ctx context.Context, w silentWait, ttl time.Duration) error
+	// Arm puts a wait's key, to live ttl: made again if the machine is sent again. The key holds when the send was.
+	Arm(ctx context.Context, w silentWait, ttl time.Duration, sent time.Time) error
+	// Held is every wait that has a key now, with when its send was.
+	Held(ctx context.Context) (map[silentWait]time.Time, error)
 	// Reported takes a wait's key away: the machine reported on the send, and nothing is to wake the controller.
 	Reported(ctx context.Context, w silentWait) error
 }
@@ -94,13 +96,13 @@ func newBusSilent(ctx context.Context, api jetstream.JetStream, silent func(sile
 	return &busSilent{kv: kv}, nil
 }
 
-func (s *busSilent) Arm(ctx context.Context, w silentWait, ttl time.Duration) error {
+func (s *busSilent) Arm(ctx context.Context, w silentWait, ttl time.Duration, sent time.Time) error {
 	key := silentKey(w)
 	// Made again, not changed: a key's time to live is set when it is made.
 	if err := s.kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
 		return err
 	}
-	_, err := s.kv.Create(ctx, key, []byte(time.Now().UTC().Format(time.RFC3339)), jetstream.KeyTTL(leastSecond(ttl)))
+	_, err := s.kv.Create(ctx, key, []byte(sent.UTC().Format(time.RFC3339Nano)), jetstream.KeyTTL(leastSecond(ttl)))
 	return err
 }
 
@@ -119,3 +121,30 @@ func leastSecond(d time.Duration) time.Duration {
 	}
 	return (d + time.Second - 1) / time.Second * time.Second
 }
+
+func (s *busSilent) Held(ctx context.Context) (map[silentWait]time.Time, error) {
+	keys, err := s.kv.Keys(ctx)
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	held := map[silentWait]time.Time{}
+	for _, key := range keys {
+		w, ok := parseSilentKey(key)
+		if !ok {
+			continue
+		}
+		e, err := s.kv.Get(ctx, key)
+		if err != nil {
+			if errors.Is(err, jetstream.ErrKeyNotFound) {
+				continue
+			}
+			return nil, err
+		}
+		at, _ := time.Parse(time.RFC3339Nano, string(e.Value()))
+		held[w] = at
+	}
+	return held, nil
+}