go/worker/keymanager: Add observer nodes automatically to ACL

.changelog/5606.feature.md

@@ -0,0 +1,5 @@
+Add observer nodes automatically to the keymanager's access list
+
+Observer nodes for a given paratime had to be added manually. This
+change brings observer nodes in line with compute nodes, which were
+added automatically.

go/worker/keymanager/watcher.go

@@ -284,6 +284,32 @@ func (w *rtNodeWatcher) watch(ctx context.Context) {
 	}
 	defer sub.Close()
 
+	// Maintain a list of allowable nodes in addition to the committee members.
+	observerNodes := map[signature.PublicKey]struct{}{}
+
+	nodeCh, nodeSub, err := w.consensus.Registry().WatchNodes(ctx)
+	if err != nil {
+		w.logger.Error("failed to subscribe to registry node updates",
+			"err", err,
+		)
+		return
+	}
+	defer nodeSub.Close()
+
+	// And populate it with the currently known set of observer nodes.
+	nodes, err := w.consensus.Registry().GetNodes(ctx, consensus.HeightLatest)
+	if err != nil {
+		w.logger.Error("failed to fetch list of nodes from the registry",
+			"err", err,
+		)
+		return
+	}
+	for _, nd := range nodes {
+		if nd.HasRoles(node.RoleObserver) && nd.HasRuntime(w.runtimeID) {
+			observerNodes[nd.ID] = struct{}{}
+		}
+	}
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -293,6 +319,11 @@ func (w *rtNodeWatcher) watch(ctx context.Context) {
 				watcher.Reset()
 				defer watcher.Freeze(0)
 
+				for sig := range observerNodes {
+					_, _ = watcher.WatchNode(ctx, sig)
+				}
+
+				// Get executor committee members.
 				cms, err := w.consensus.Scheduler().GetCommittees(ctx, &scheduler.GetCommitteesRequest{
 					Height:    consensus.HeightLatest,
 					RuntimeID: w.runtimeID,
@@ -321,6 +352,12 @@ func (w *rtNodeWatcher) watch(ctx context.Context) {
 				// nodes have been set (even if the new set is empty).
 				continue
 			}
+		case ne := <-nodeCh:
+			if ne.IsRegistration && ne.Node.HasRoles(node.RoleObserver) && ne.Node.HasRuntime(w.runtimeID) {
+				observerNodes[ne.Node.ID] = struct{}{}
+			} else {
+				delete(observerNodes, ne.Node.ID)
+			}
 		}
 
 		w.accessList.UpdateNodes(w.runtimeID, watcher.GetNodes())