.

wcfs/wcfs.go

@@ -1608,13 +1608,11 @@ func (wlink *WatchLink) _serve() (err error) {
 func (wlink *WatchLink) handleWatch(ctx context.Context, stream uint64, msg string) (err error) {
 	defer xerr.Contextf(&err, "%d", stream)
 
-	// FIXME error "at is too far away" from setupWatch shuold be not fatal
-	// (XXX bad parse watch -> fatal (not following protocol))
-	// XXX kill fatal replies? (i.e. make almost everything non-fatal ?)
 	err = wlink._handleWatch(ctx, msg)
 	reply := "ok"
 	if err != nil {
 		reply = fmt.Sprintf("error %s", err)
+		err = nil
 	}
 
 	err2 := wlink.send(ctx, stream, reply)

wcfs/wcfs_test.py

@@ -477,8 +477,8 @@ class tWatchLink:
         t._wrx = os.fdopen(wh, 'rb')
         t._wtx = os.fdopen(wh2, 'wb')
 
-        # .rx_eof becomes ready when wcfs closes its tx side
-        t.rx_eof = chan()
+        t.rx_eof = chan()   # .rx_eof becomes ready when wcfs closes its tx side
+        t.fatalv = []       # fatal messages go here
 
         # inv.protocol message IO
         t._acceptq  = chan() # (stream, msg)    server originated messages go here
@@ -555,6 +555,7 @@ class tWatchLink:
             if stream == 0: # control/fatal message from wcfs
                 # XXX print -> receive somewhere?
                 print('C: watch  : rx fatal: %r' % msg)
+                t.fatalv.append(msg)
                 continue
 
             reply = bool(stream % 2)
@@ -586,8 +587,11 @@ class tWatchLink:
     # XXX +ctx?
     def _send(t, stream, msg):
         assert '\n' not in msg
+        pkt = b"%d %s\n" % (stream, msg)
+        t._write(pkt)
+
+    def _write(t, pkt):
         with t._txmu:
-            pkt = b"%d %s\n" % (stream, msg)
             #print('C: watch  : tx: %r' % pkt)
             t._wtx.write(pkt)
             t._wtx.flush()
@@ -886,21 +890,22 @@ def test_wcfs():
     # >>> invalidation protocol
     print('\n\n inv. protocol \n\n')
 
-    # invalid requests -> wcfs replies error
-
-    # XXX invalid request not following frame structure `<stream> ...`
 
+    # invalid requests -> wcfs replies error
     wl = t.openwatch()
     assert wl.sendReq(context.background(), b'bla bla') ==  \
             b'error bad watch: not a watch request: "bla bla"'
 
-    # wcfs must close watch link after invalid request
+    # invalid request not following frame structure -> fatal + wcfs must close watch link
+    assert wl.fatalv == []
+    wl._write(b'zzz hello\n')
     _, _rx = select(
         time.after(3*time.second).recv,
         wl.rx_eof.recv,
     )
     if _ == 0:
-        raise RuntimeError("%s: did not rx EOF after invalid watch request" % wl)
+        raise RuntimeError("%s: did not rx EOF after bad frame " % wl)
+    assert wl.fatalv == []  # XXX
 
     wl.close()
     return