.

wcfs/wcfs.go

@@ -1458,6 +1458,8 @@ func (wlink *WatchLink) _serveRX() (err error) {
 	defer xerr.Contextf(&err, "wlink %d: serve rx", wlink.id)
 	r := bufio.NewReader(wlink.sk)
 
+	// XXX close .sk on error/wcfs stopping
+
 	// XXX write to peer if it was logical error on client side
 	// XXX on which stream? -1?
 
@@ -1490,30 +1492,43 @@ func (wlink *WatchLink) _serveRX() (err error) {
 		}
 
 		// client-initiated watch request
-		foid, at, err := parseWatch(msg)
+		err = wlink.handleSetupWatch(stream, msg)
 		if err != nil {
-			return fmt.Errorf("%d: %s", stream, err)
+			panic(err)	// XXX
 		}
+	}
+}
 
-		ctx := context.TODO()	// XXX ctx = ?
+// handleSetupWatch handles watch request from client.
+//
+// returned error comes without full error prefix.
+func (wlink *WatchLink) handleSetupWatch(stream uint64, msg string) (err error) {
+	defer xerr.Contextf(&err, "%d", stream)
 
-		fmt.Printf("S: watch: AAA\n")
+	foid, at, err := parseWatch(msg)
+	if err != nil {
+		return err
+	}
 
-		err = wlink.setupWatch(ctx, foid, at)
-		if err != nil {
-			fmt.Printf("S: watch: QQQ: %s\n", err)
-			return fmt.Errorf("%d: %s", stream, err)
-		}
+	ctx := context.TODO()	// XXX ctx = ?
 
-		fmt.Printf("S: watch: BBB\n")
+	fmt.Printf("S: watch: AAA\n")
 
-		err = wlink.send(ctx, stream, "ok")
-		if err != nil {
-			panic(err)	// XXX
-		}
+	err = wlink.setupWatch(ctx, foid, at)
+	if err != nil {
+		fmt.Printf("S: watch: QQQ: %s\n", err)
+		return err
+	}
 
-		fmt.Printf("S: watch: CCC\n")
+	fmt.Printf("S: watch: BBB\n")
+
+	err = wlink.send(ctx, stream, "ok")
+	if err != nil {
+		return err
 	}
+
+	fmt.Printf("S: watch: CCC\n")
+	return nil
 }
 
 // sendReq sends wcfs-originated request to client and returns client response.
@@ -1533,6 +1548,7 @@ func (wlink *WatchLink) sendReq(ctx context.Context, req string) (reply string,
 
 	select {
 	case <-ctx.Done():
+		// XXX del rxTab[stream] ?
 		return "", ctx.Err()
 
 	case reply = <-rxq:

wcfs/wcfs_test.py

@@ -443,7 +443,7 @@ class tWatch:
         with t._txmu:
             pkt = b"%d %s\n" % (stream, msg)
             print('C: watch: tx: %r' % pkt)
-            t.wtx.write(pkt)   # XXX read/write don't work in parallel?
+            t.wtx.write(pkt)
             t.wtx.flush()
 
     # sendReq sends client -> server request and returns server reply.
@@ -674,6 +674,7 @@ def test_wcfs():
     # XXX read file[blk]=hole; then file[blk]=zblk - must be invalidated and
     # setupWatch must send pins.
 
+    # XXX watch @at when file did not existed
 
 
 def test_wcfs_invproto():