.

go/neo/client.go

@@ -115,7 +115,7 @@ func (c *Client) Run(ctx context.Context) (err error) {
 		ctx, cancel := xcontext.Merge/*Cancel*/(ctx, runCtx)
 		defer cancel()
 
-		return c.node.TalkMaster(ctx, func(ctx context.Context, mlink *neonet.NodeLink) error {
+		return c.node.TalkMaster(ctx, func(ctx context.Context, mlink *_MasterLink) error {
 			// XXX errctx  ("on redial"? "connected"?)
 
 			c.head0 = c.head
@@ -124,7 +124,7 @@ func (c *Client) Run(ctx context.Context) (err error) {
 
 			// launch master notifications receiver
 			wg.Go(func(ctx context.Context) error {
-				return c.recvMaster(ctx)
+				return c.recvMaster(ctx, mlink)
 			})
 
 			// sync lastTid with master
@@ -142,11 +142,11 @@ func (c *Client) Run(ctx context.Context) (err error) {
 }
 
 // recvMaster receives and handles notifications from master.
-func (c *Client) recvMaster(ctx context.Context) (err error) {
+func (c *Client) recvMaster(ctx context.Context, mlink *_MasterLink) (err error) {
 	defer task.Running(&ctx, "rx")(&err)
 
 	for {
-		req, err := c.node.RecvM1()
+		req, err := mlink.Recv1()
 		if err != nil {
 			return err
 		}
@@ -207,7 +207,7 @@ func (c *Client) invalidateObjects(msg *proto.InvalidateObjects) error {
 }
 
 // syncMaster asks M for DB head right after identification.
-func (c *Client) syncMaster(ctx context.Context, mlink *neonet.NodeLink) (err error) {
+func (c *Client) syncMaster(ctx context.Context, mlink *_MasterLink) (err error) {
 	defer task.Running(&ctx, "sync0")(&err) // XXX unify with Sync ?
 
 	// query last_tid

go/neo/mastered.go

@@ -58,6 +58,8 @@ import (
 //
 // This pipeline is operated by TalkMaster.
 // The connection to master is persisted by redial as needed.
+//
+// XXX update after introduction of _MasterLink
 type _MasteredNode struct {
 	myInfo      proto.NodeInfo // type, laddr, nid, state, idtime
 	ClusterName string
@@ -81,6 +83,16 @@ type _MasteredNode struct {
 				    // XXX just use `.myInfo.NodeType == STORAGE` instead?
 }
 
+// _MasterLink represents NodeLink to master with Recv1 filtered through ^^^ XXX
+// XXX place
+type _MasterLink struct {
+	*neonet.NodeLink
+	node *_MasteredNode
+}
+func (mlink *_MasterLink) Recv1() (neonet.Request, error) {
+	return mlink.node.recvM1()
+}
+
 // _RxM represents a request or event received from master.
 type _RxM struct {
 	Req neonet.Request
@@ -137,7 +149,7 @@ func newMasteredNode(typ proto.NodeType, clusterName string, net xnet.Networker,
 // See top-level _MasteredNode overview for details.
 //
 // XXX -> FollowMaster? AdhereMaster?
-func (node *_MasteredNode) TalkMaster(ctx context.Context, f func(context.Context, *neonet.NodeLink) error) (err error) {
+func (node *_MasteredNode) TalkMaster(ctx context.Context, f func(context.Context, *_MasterLink) error) (err error) {
 	// me0 describes local node when it starts connecting to master, e.g. 'client C?'.
 	// we don't use just NID because it is initially 0 and because master can tell us to change it.
 	me0 := strings.ToLower(node.myInfo.Type.String())
@@ -176,7 +188,7 @@ func (node *_MasteredNode) TalkMaster(ctx context.Context, f func(context.Contex
 	}
 }
 
-func (node *_MasteredNode) talkMaster1(ctx, ctxPreTalkM context.Context, f func(context.Context, *neonet.NodeLink) error) error {
+func (node *_MasteredNode) talkMaster1(ctx, ctxPreTalkM context.Context, f func(context.Context, *_MasterLink) error) error {
 	reqID := &proto.RequestIdentification{
 		NodeType:	node.myInfo.Type,
 		NID:		node.myInfo.NID,
@@ -259,7 +271,7 @@ func (node *_MasteredNode) talkMaster1(ctx, ctxPreTalkM context.Context, f func(
 		})
 		// run user code
 		wg.Go(func(ctx context.Context) error {
-			return f(ctx, mlink)
+			return f(ctx, &_MasterLink{mlink, node})
 		})
 
 		return wg.Wait()
@@ -307,10 +319,10 @@ func (node *_MasteredNode) recvMaster1(ctx context.Context, req neonet.Request)
 
 var errMasterDisconect = errors.New("master disconnected")
 
-// RecvM1 receives request from master filtered through δstate handler.
+// recvM1 receives request from master filtered through δstate handler.
 //
 // Must be called only when master link is established - e.g. from under TalkMaster.
-func (node *_MasteredNode) RecvM1() (neonet.Request, error) {
+func (node *_MasteredNode) recvM1() (neonet.Request, error) {
 	rx, ok := <-node.rxm
 	if !ok {
 		return neonet.Request{}, errMasterDisconect

go/neo/storage.go

@@ -100,7 +100,7 @@ func (stor *Storage) Run(ctx context.Context, l xnet.Listener) (err error) {
 
 	// connect to master and get commands and updates from it
 	wg.Go(func(ctx context.Context) error {
-		return stor.node.TalkMaster(ctx, func(ctx context.Context, mlink *neonet.NodeLink) error {
+		return stor.node.TalkMaster(ctx, func(ctx context.Context, mlink *_MasterLink) error {
 			// XXX move -> SetNumReplicas handler
 			// // NumReplicas: neo/py meaning for n(replica) = `n(real-replica) - 1`
 			// if !(accept.NumPartitions == 1 && accept.NumReplicas == 0) {
@@ -114,7 +114,7 @@ func (stor *Storage) Run(ctx context.Context, l xnet.Listener) (err error) {
 			}
 
 			// we got StartOperation command. Let master drive us during service phase.
-			return stor.m1serve(ctx, reqStart)
+			return stor.m1serve(ctx, mlink, reqStart)
 		})
 	})
 
@@ -172,11 +172,11 @@ func (stor *Storage) Run(ctx context.Context, l xnet.Listener) (err error) {
 // return error indicates:
 // - nil:  initialization was ok and a command came from master to start operation.
 // - !nil: initialization was cancelled or failed somehow.
-func (stor *Storage) m1initialize(ctx context.Context, mlink *neonet.NodeLink) (reqStart *neonet.Request, err error) {
-	defer task.Runningf(&ctx, "init %v", mlink)(&err)
+func (stor *Storage) m1initialize(ctx context.Context, mlink *_MasterLink) (reqStart *neonet.Request, err error) {
+	defer task.Runningf(&ctx, "init %s", mlink)(&err)
 
 	for {
-		req, err := stor.node.RecvM1()
+		req, err := mlink.Recv1()
 		if err != nil {
 			return nil, err
 		}
@@ -253,9 +253,8 @@ func (stor *Storage) m1initialize1(ctx context.Context, req neonet.Request) erro
 // it always returns with an error describing why serve had to be stopped -
 // either due to master commanding us to stop, or context cancel or some other
 // error.
-func (stor *Storage) m1serve(ctx context.Context, reqStart *neonet.Request) (err error) {
-	mlink := reqStart.Link()
-	defer task.Runningf(&ctx, "serve %v", mlink)(&err)
+func (stor *Storage) m1serve(ctx context.Context, mlink *_MasterLink, reqStart *neonet.Request) (err error) {
+	defer task.Runningf(&ctx, "serve %s", mlink)(&err)
 
 	// refresh stor.opCtx and cancel it when we finish so that client
 	// handlers know they need to stop operating as master told us to do so.
@@ -275,7 +274,7 @@ func (stor *Storage) m1serve(ctx context.Context, reqStart *neonet.Request) (err
 
 	for {
 		// XXX abort on ctx (XXX or upper?)
-		req, err := stor.node.RecvM1()
+		req, err := mlink.Recv1()
 		if err != nil {
 			return err
 		}