.

go/neo/master.go

@@ -33,6 +33,7 @@ import (
 	"sync"
 	"time"
 
+	xxcontext "lab.nexedi.com/kirr/go123/xcontext"
 	"lab.nexedi.com/kirr/go123/xnet"
 	"lab.nexedi.com/kirr/go123/xsync"
 
@@ -378,6 +379,7 @@ loop:
 			// FIXME if identify=ok -> subscribe to δ(nodeTab) and send initial nodeTab right after accept  (accept should do it?)
 			node, state0, resp := m.identify(ctx, n, /* XXX only accept storages -> PENDING */)
 
+			// XXX -> move to identify (spawn under peerWork.wg) ?
 			if node == nil {
 				goreject(ctx, wg, n.req, resp)
 				break
@@ -389,15 +391,17 @@ loop:
 			go func() {
 				defer wg.Done()
 
-/*
 				// start recovery
 				var pt *xneo.PartitionTable
-				err := acceptAndRun(func(node *_PeerNode) error {
+				err := acceptAndRun(func(workCtx context.Context, node *xneo.PeerNode) error {
+					ctx, cancel := xxcontext.Merge/*Cancel*/(ctx, workCtx)
+					defer cancel()
 					pt, err = storCtlRecovery(ctx, node)
+					return err
 				})
-				recevery <- storRecovery{stor: node, partTab: pt, err: err}
-*/
+				recovery <- storRecovery{stor: node, partTab: pt, err: err}
 
+/*
 				err := m.accept(node, state0, n.req, resp)
 				if err != nil {
 					recovery <- storRecovery{stor: node, err: err}
@@ -406,6 +410,7 @@ loop:
 
 				// start recovery
 				storCtlRecovery(ctx, node, recovery)
+*/
 			}()
 
 		// XXX <-m.nodeLeave
@@ -1266,10 +1271,11 @@ func (m *Master) accept(peer *xneo.PeerNode, state0 *xneo.ClusterStateSnapshot,
 
 	// XXX send clusterState too? (NEO/py does not send it)
 
+	var w *peerWork // XXX stub <- = .peerWorkTab[peer.NID] set from main
 
 	// go proxy δstate ... XXX
 	// XXX under which wg? -> under per-peer wg
-	wg.Go(func(ctx context.Context) (err error) {
+	w.wg.Go(func(ctx context.Context) (err error) {
 		defer task.Runningf(&ctx, "send cluster updates")(&err)
 
 		stateCode := state0.Code
@@ -1281,7 +1287,7 @@ func (m *Master) accept(peer *xneo.PeerNode, state0 *xneo.ClusterStateSnapshot,
 			case <-ctx.Done():
 				return ctx.Err()
 
-			case δstate = <-notifyq:	// XXX could be also closed?
+			case δstate = <-w.notifyq:	// XXX could be also closed?
 			}
 
 			var msg proto.Msg