wcfs_test.py

# -*- coding: utf-8 -*-
# Copyright (C) 2018-2019  Nexedi SA and Contributors.
#                          Kirill Smelkov <kirr@nexedi.com>
#
# This program is free software: you can Use, Study, Modify and Redistribute
# it under the terms of the GNU General Public License version 3, or (at your
# option) any later version, as published by the Free Software Foundation.
#
# You can also Link and Combine this program with other software covered by
# the terms of any of the Free Software licenses or any of the Open Source
# Initiative approved licenses and Convey the resulting work. Corresponding
# source of such a combination shall include the source code for all other
# software used.
#
# This program is distributed WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options.
"""wcfs_test tests wcfs filesystem from outside as python client process"""

from __future__ import print_function

from wendelin.lib.testing import getTestDB
from wendelin.lib.zodb import dbclose
from wendelin.lib.mem import memcpy
from wendelin.bigfile.file_zodb import ZBigFile
from wendelin.bigfile.tests.test_filezodb import blksize
from wendelin import wcfs

import transaction
from persistent import Persistent
from persistent.timestamp import TimeStamp

import os, os.path, subprocess, threading
from errno import EINVAL
from golang import chan, func, defer, select
from golang import context, sync
from zodbtools.util import ashex as h, fromhex
from pytest import raises
from six import reraise
from .internal import mm

# setup:
# - create test database, compute zurl and mountpoint for wcfs
# - at every test: make sure wcfs is not running before & after the test.

testdb = None
testzurl = None     # URL of testdb
testmntpt = None    # wcfs is mounted here
def setup_module():
    global testdb, testzurl, testmntpt
    testdb = getTestDB()
    testdb.setup()

    zstor = testdb.getZODBStorage()
    testzurl = wcfs.zstor_2zurl(zstor)
    zstor.close()
    testmntpt = wcfs._mntpt_4zurl(testzurl)
    os.rmdir(testmntpt)

def teardown_module():
    testdb.teardown()

# make sure we start every test without wcfs server running.
def setup_function(f):
    assert not os.path.exists(testmntpt)

# make sure we unmount wcfs after every test.
def teardown_function(f):
    mounted = not subprocess.call(["mountpoint", "-q", testmntpt])
    if mounted:
        subprocess.check_call(["fusermount", "-u", testmntpt])
    if os.path.exists(testmntpt):
        os.rmdir(testmntpt)


# ---- test join/autostart ----

# test that zurl does not change from one open to another storage open.
def test_zurlstable():
    for i in range(10):
        zstor = testdb.getZODBStorage()
        zurl  = wcfs.zstor_2zurl(zstor)
        zstor.close()
        assert zurl == testzurl

# test that join works.
@func
def test_join():
    zurl = testzurl
    with raises(RuntimeError, match="wcfs: join .*: server not started"):
        wcfs.join(zurl, autostart=False)

    wc = wcfs._start(zurl)
    defer(wc.close)
    assert wc.mountpoint == testmntpt
    assert readfile(wc.mountpoint + "/.wcfs/zurl") == zurl
    assert os.path.isdir(wc.mountpoint + "/head")
    assert os.path.isdir(wc.mountpoint + "/head/bigfile")

    wc2 = wcfs.join(zurl, autostart=False)
    defer(wc2.close)
    assert wc2.mountpoint == wc.mountpoint

# test that join(autostart=y) works.
@func
def test_join_autostart():
    zurl = testzurl
    with raises(RuntimeError, match="wcfs: join .*: server not started"):
        wcfs.join(zurl, autostart=False)

    wc = wcfs.join(zurl, autostart=True)
    defer(wc.close)
    assert wc.mountpoint == testmntpt
    assert readfile(wc.mountpoint + "/.wcfs/zurl") == zurl
    assert os.path.isdir(wc.mountpoint + "/head")
    assert os.path.isdir(wc.mountpoint + "/head/bigfile")


# --- test access to data ----

# DF represents a change in files space.
# it corresponds to ΔF in wcfs.go .
class DF:
    # .rev      tid
    # .byfile   {} ZBigFile -> DFile
    def __init__(dF):
        # rev set from outside
        dF.byfile = {}

# DFile represents a change to one file.
# it is is similar to ΔFile in wcfs.go .
class DFile:
    # .rev      tid
    # .ddata    {} blk -> data      XXX name
    def __init__(dfile):
        # rev set from outside
        dfile.ddata = {}

# tDB provides database/wcfs testing environment.
#
# BigFiles opened under tDB are represented as tFile - see .open for details.
# Watches opened under tDB are represented as tWatch - see .openwatch for details.
#
# XXX .open         -> .topen
# XXX .openwatch    -> .topenwatch ?
#
# XXX print -> t.trace/debug() + t.verbose depending on py.test -v -v ?
class tDB:
    def __init__(t):
        t.root  = testdb.dbopen()
        t.wc    = wcfs.join(testzurl, autostart=True)

        # ZBigFile(s) scheduled for commit
        t._changed = {} # ZBigFile -> {} blk -> data

        # committed: head + δF history
        t.head   = None
        t._headv = [] # XXX -> just use dFtail[·].rev ?
        t.dFtail = [] # of DF

        # fh(.wcfs/zhead) + history of zhead read from there
        t._wc_zheadfh = open(t.wc.mountpoint + "/.wcfs/zhead")
        t._wc_zheadv  = []

        # tracked tFiles & tWatches
        t._tracked = set()

    # close closes test database as well as all tracked files, watches and wcfs.
    def close(t):
        for tf in t._tracked.copy():
            tf.close()
        assert len(t._tracked) == 0
        t._wc_zheadfh.close()
        t.wc.close()
        dbclose(t.root)

    # open opens wcfs file corresponding to zf@at and starts to track it.
    # see returned tFile for details.
    def open(t, zf, at=None):   # -> tFile
        return tFile(t, zf, at=at)

    # openwatch opens /head/watch on wcfs.
    # see returned tWatch for details.
    def openwatch(t):   # -> tWatch
        return tWatch(t)

    # change schedules zf to be changed according changeDelta at commit.
    #
    # changeDelta is {} blk -> data.
    def change(t, zf, changeDelta):
        assert isinstance(zf, ZBigFile)
        zfDelta = t._changed.setdefault(zf, {})
        for blk, data in changeDelta.iteritems():
            assert len(data) <= zf.blksize
            zfDelta[blk] = data

    # commit commits transaction and makes sure wcfs is synchronized to it.
    #
    # It remembers/returns committed transaction ID.
    def commit(t):
        # perform modifications scheduled by change.
        # use !wcfs mode so that we prepare data independently of wcfs code paths.
        dF = DF()
        for zf, zfDelta in t._changed.items():
            dfile = DFile()
            zfh = zf.fileh_open(_use_wcfs=False)
            for blk, data in zfDelta.iteritems():
                dfile.ddata[blk] = data
                data += b'\0'*(zf.blksize - len(data))  # trailing \0
                vma = zfh.mmap(blk, 1)
                memcpy(vma, data)
            dF.byfile[zf] = dfile

        # NOTE there is no clean way to retrieve tid of just committed transaction
        #      we use last._p_serial as workaround.
        t.root['_last'] = last = Persistent()
        last._p_changed = 1

        transaction.commit()
        head = last._p_serial
        t.head = head
        t._headv.append(head)

        print('\nM: commit -> %s' % t.hat(head))
        for zf, zfDelta in t._changed.items():
            print('M:      f<%s>\t%s' % (h(zf._p_oid), sorted(zfDelta.keys())))
        t._changed = {}

        dF.rev = head
        for dfile in dF.byfile.values():
            dfile.rev = head
        t.dFtail.append(dF)

        # synchronize wcfs to db
        t._wcsync()

        return head

    # _wcsync makes sure wcfs is synchronized to latest committed transaction.
    def _wcsync(t):
        while len(t._wc_zheadv) < len(t._headv):
            l = t._wc_zheadfh.readline()
            #print('> zhead read: %r' % l)
            l = l.rstrip('\n')
            wchead = fromhex(l)
            i = len(t._wc_zheadv)
            if wchead != t._headv[i]:
                raise RuntimeError("wcsync #%d: wczhead (%s) != zhead (%s)" % (i, h(wchead), h(t._headv[i])))
            t._wc_zheadv.append(wchead)

        # head/at = last txn of whole db
        assert t.read("head/at") == h(t.head)


    # path returns path for object on wcfs.
    # - str:        wcfs root + obj;
    # - Persistent: wcfs root + (head|@<at>)/bigfile/obj
    def path(t, obj, at=None):
        if isinstance(obj, Persistent):
            head = "head/" if at is None else ("@%s/" % h(at))
            obj  = "%s/bigfile/%s" % (head, h(obj._p_oid))
            at   = None

        assert isinstance(obj, str)
        assert at is None  # must not be used with str

        return os.path.join(t.wc.mountpoint, obj)

    # read reads file corresponding to obj on wcfs.
    def read(t, obj, at=None):
        path = t.path(obj, at=at)
        return readfile(path)

    # stat stats file corresponding to obj on wcfs.
    def stat(t, obj, at=None):
        path = t.path(obj, at=at)
        return os.stat(path)

    # _open opens file corresponding to obj on wcfs.
    def _open(t, obj, mode='rb', at=None):
        path = t.path(obj, at=at)
        return open(path, mode, 0)  # unbuffered


    # hat returns string for at.
    # it gives both symbolic version and raw hex for at, for example:
    #   @at2 (03cf7850500b5f66)
    def hat(t, at):
        try:
            i = t._headv.index(at)
        except ValueError:
            return "@" + h(at)

        return "@at%d (%s)" % (i, h(at))


    # zfiles returns ZBigFiles that were ever changed under t.
    def zfiles(t):
        zfs = set()
        for dF in t.dFtail:
            for zf in dF.byfile:
                if zf not in zfs:
                    zfs.add(zf)
        return zfs


# tFile provides testing environment for one bigfile on wcfs.
#
# .blk() provides access to data of a block. .cached() gives state of which
# blocks are in OS pagecache. .assertCache and .assertBlk/.assertData assert
# on state of cache and data.
class tFile:
    # maximum number of pages we mmap for 1 file.
    # this should be not big not to exceed mlock limit.
    _max_tracked = 8

    def __init__(t, tdb, zf, at=None):
        assert isinstance(zf, ZBigFile)
        t.tdb = tdb
        t.f   = tdb._open(zf, at=at)
        t.blksize = zf.blksize

        # mmap the file past the end up to _max_tracked pages and lock the
        # pages with MLOCK_ONFAULT. This way when a page is read by mmap access
        # we have the guarantee from kernel that the page will stay in
        # pagecache. We rely on this to verify OS cache state.
        assert t.blksize % mm.PAGE_SIZE == 0
        t.fmmap = mm.map_ro(t.f.fileno(), 0, t._max_tracked*t.blksize)
        mm.lock(t.fmmap, mm.MLOCK_ONFAULT)

        tdb._tracked.add(t)

    def close(t):
        t.tdb._tracked.remove(t)
        mm.unmap(t.fmmap)
        t.f.close()

    # blk returns bytearray view of file[blk].
    def blk(t, blk):
        assert blk <= t._max_tracked
        return bytearray(t.fmmap[blk*t.blksize:(blk+1)*t.blksize])

    # cached returns [] with indicating whether a file block is cached or not.
    # 1 - cached, 0 - not cached, fractional (0,1) - some pages of the block are cached some not.
    def cached(t):
        l = t._sizeinblk()
        incorev = mm.incore(t.fmmap[:l*t.blksize])
        # incorev is in pages; convert to in blocks
        assert t.blksize % mm.PAGE_SIZE == 0
        blkpages = t.blksize // mm.PAGE_SIZE
        cachev = [0.]*l
        for i, v in enumerate(incorev):
            blk = i // blkpages
            cachev[blk] += bool(v)
        for blk in range(l):
            cachev[blk] /= blkpages
            if cachev[blk] == int(cachev[blk]):
                cachev[blk] = int(cachev[blk])  # 0.0 -> 0, 1.0 -> 1
        return cachev

    # _sizeinblk returns file size in blocks.
    def _sizeinblk(t):
        st = os.fstat(t.f.fileno())
        assert st.st_size % t.blksize == 0
        assert st.st_size // t.blksize <= t._max_tracked
        return st.st_size // t.blksize

    # assertCache asserts state of OS cache for file.
    #
    # incorev is [] of 1/0 representing whether block data is present or not.
    def assertCache(t, incorev):
        assert t.cached() == incorev

    # assertBlk asserts that file block #blk has data as expected.
    #
    # Expected data may be given with size < t.blksize. In such case the data
    # is implicitly appended with trailing zeros.
    def assertBlk(t, blk, data):
        assert len(data) <= t.blksize
        data += b'\0'*(t.blksize - len(data))   # tailing zeros
        assert blk < t._sizeinblk()

        # XXX assert individually for every block's page? (easier debugging?)
        assert t.blk(blk) == data, ("#blk: %d" % blk)

        # we just accessed the block - it has to be in OS cache
        assert t.cached()[blk] == 1

    # assertData asserts that file has data blocks as specified.
    #
    # Expected blocks may be given with size < zf.blksize. In such case they
    # are implicitly appended with trailing zeros.
    #
    # It also checks file size and optionally mtime.
    def assertData(t, datav, mtime=None):
        st = os.fstat(t.f.fileno())
        assert st.st_size == len(datav)*t.blksize
        if mtime is not None:
            assert st.st_mtime == tidtime(mtime)

        for blk, data in enumerate(datav):
            t.assertBlk(blk, data)

        # all blocks must be in cache after we touched them all
        t.assertCache([1]*len(datav))


# tWatch provides testing environment for /head/watch opened on wcfs.
#
# .sendReq()/.recvReq() provides raw IO in terms of wcfs invalidation protocol messages.
# .watch() setups a watch for a file and verifies ... XXX
class tWatch:

    def __init__(t, tdb):
        t.tdb = tdb

        # head/watch handle.
        #
        # python/stdio lock file object on read/write, however we need both
        # read and write to be working simultaneously.
        # -> use 2 separate file objects for rx and tx.
        #
        # fdopen takes ownership of file descriptor and closes it when file
        # object is closed -> dup fd so that each file object has its own fd.
        wh  = os.open(tdb.path("head/watch"), os.O_RDWR)
        wh2 = os.dup(wh)
        t.wrx = os.fdopen(wh, 'rb')
        t.wtx = os.fdopen(wh2, 'wb')

        # inv.protocol message IO
        t._acceptq  = chan() # (stream, msg)    server originated messages go here
        t._rxmu     = threading.Lock()
        t._rxtab    = {}     # stream -> rxq    server replies go via here
        t._accepted = set()  # of stream        streams we accepted but did not replied yet

        t._txmu     = threading.Lock()  # serializes writes

        serveCtx, t._serveCancel = context.with_cancel(context.background())
        t._serveWG = sync.WorkGroup(serveCtx)
        t._serveWG.go(t._serveRecv)

        # this tWatch currently watches files at particular state.
        t._watching = {}    # {} ZBigFile -> @at

        tdb._tracked.add(t)

    def close(t):
        t.tdb._tracked.remove(t)

        t._serveCancel()
        # ask wcfs to close its tx & rx sides; close(wcfs.tx) wakes up
        # _serveRecv on client (= on us).
        t._send(1, b'bye')
        # XXX we can get stuck here if wcfs does not behave as we want.
        # XXX in particular if there is a silly - e.g. syntax or type error in
        #     test code - we curently get stuck here.
        #
        # XXX -> better pthread_kill(SIGINT) instead of relying on wcfs proper behaviour?
        try:
            t._serveWG.wait()
        except Exception as e:
            # canceled is expected and ok
            if e != context.canceled:
                reraise(e, None, e.__traceback__)

        t.wtx.close()
        t.wrx.close()

        # wakeup everyone waiting for rx
        t._acceptq.close()
        with t._rxmu:
            rxtab = t._rxtab
            t._rxtab = None     # don't allow new rxtab registers
        for rxq in rxtab.values():
            rxq.close()


    # ---- message IO ----

    # _serveRecv receives messages from .w and dispatches them according to streamID.
    @func
    def _serveRecv(t, ctx):
        while 1:
            # NOTE: .close() makes sure .wrx.read*() will wake up
            l = t.wrx.readline()
            #print('C: watch  : rx: %r' % l)
            if len(l) == 0:
                break   # closed

            # <stream> ... \n
            stream, msg = l.split(' ', 1)
            stream = int(stream)
            msg = msg.rstrip('\n')

            reply = bool(stream % 2)
            if reply:
                with t._rxmu:
                    assert stream in t._rxtab
                    rxq = t._rxtab.pop(stream)
                _, _rx = select(
                    ctx.done().recv,    # 0
                    (rxq.send, msg),    # 1
                )
                if _ == 0:
                    raise ctx.err()
            else:
                with t._rxmu:
                    assert stream not in t._accepted
                    t._accepted.add(stream)
                _, _rx = select(
                    ctx.done().recv,                    # 0
                    (t._acceptq.send, (stream, msg)),   # 1
                )
                if _ == 0:
                    raise ctx.err()


    # _send sends raw message via specified stream.
    #
    # multiple _send can be called in parallel - _send serializes writes.
    # XXX +ctx?
    def _send(t, stream, msg):
        assert '\n' not in msg
        with t._txmu:
            pkt = b"%d %s\n" % (stream, msg)
            #print('C: watch  : tx: %r' % pkt)
            t.wtx.write(pkt)
            t.wtx.flush()

    # sendReq sends client -> server request and returns server reply.
    #
    # only 1 sendReq must be used at a time.    # XXX relax?
    def sendReq(t, ctx, req):
        stream = 1

        rxq = chan()
        with t._rxmu:
            assert stream not in t._rxtab
            t._rxtab[stream] = rxq

        t._send(stream, req)

        _, _rx = select(
            ctx.done().recv,    # 0
            rxq.recv,           # 1
        )
        if _ == 0:
            raise ctx.err()
        return _rx

    # recvReq receives client <- server request.
    #
    # multiple recvReq could be used at a time.
    def recvReq(t, ctx): # -> tSrvReq | None when EOF
        _, _rx = select(
            ctx.done().recv,    # 0
            t._acceptq.recv,    # 1
        )
        if _ == 0:
            raise ctx.err()

        rx = _rx
        if rx is None:
            return rx

        stream, msg = rx
        return tSrvReq(t, stream, msg)


# tSrvReq represents 1 server-initiated wcfs request received over /head/watch.
class tSrvReq:
    def __init__(req, twatch, stream, msg):
        req.twatch = twatch
        req.stream = stream
        req.msg    = msg

    def reply(req, answer):
        #print('C: reply %s <- %r ...' % (req, answer))
        t = req.twatch
        with t._rxmu:
            assert req.stream in t._accepted

        t._send(req.stream, answer)

        with t._rxmu:
            assert req.stream in t._accepted
            t._accepted.remove(req.stream)

        # XXX also track as answered? (and don't accept with the same ID ?)


# ---- watch setup/adjust ----

# _pinAt returns which blocks needs to be pinned for zf@at.
#
# it does not take into account whether blocks are in cache or not and computes
# pin from all changes.
@func(tWatch)
def _pinAt(w, zf, at):  # -> pin = {} blk -> rev
    t = w.tdb

    # all changes to zf
    vdf = [_.byfile[zf] for _ in t.dFtail if zf in _.byfile]

    # {} blk -> at for changes ∈ (at, head]
    pin = {}
    for df in [_ for _ in vdf if _.rev > at]:
        for blk in df.ddata:
            if blk in pin:
                continue
            # history of blk changes <= at
            blkhistoryat = [_.rev for _ in vdf if blk in _.ddata and _.rev <= at]
            if len(blkhistoryat) == 0:
                pinrev = t._headv[0]    # was hole - at0  XXX -> pin to @00?
            else:
                pinrev = max(blkhistoryat)
            assert pinrev <= at
            pin[blk] = pinrev

    return pin

# watch sets up a watch for file@at.
# XXX and verifies that wcfs sends correct initial pins?
# XXX or adjusts
@func(tWatch)
def watch(w, zf, at):   # XXX -> ?
    t = w.tdb
    at_prev = w._watching.get(zf)   # we were previously watching zf @at_prev
    at_from = ''
    if at_prev is not None:
        at_from = '(%s ->) ' % t.hat(at_prev)
    print('\nC: setup watch f<%s> %s%s' % (h(zf._p_oid), at_from, t.hat(at)))

    # pinstr returns human-readable representation for {}blk->rev
    def pinstr(pin):
        pinv = ['%d: %s' % (blk, t.hat(pin[blk])) for blk in sorted(pin.keys())]
        return '{%s}' % ', '.join(pinv)


    pin_prev = {}
    if at_prev is not None:
        assert at_prev <= at, 'TODO %s -> %s' % (t.hat(at_prev), t.hat(at))
        pin_prev = w._pinAt(zf, at_prev)

    pin = w._pinAt(zf, at)

    if at_prev != at and at_prev is not None:
        print('\n%s\n%s' % (pinstr(pin_prev), pinstr(pin)))

    for blk in set(pin_prev.keys()).union(pin.keys()):
        # blk ∉ pin_prev,   blk ∉ pin       -> cannot happen
        assert (blk in pin_prev) or (blk in pin)

        # blk ∉ pin_prev,   blk ∈ pin       -> cannot happen
        if blk not in pin_prev and blk in pin:
            if at_prev is not None:
                assert False, '#%d pinned %s; not pinned %s' % (t.hat(at_prev), t.hat(at))

        # blk ∈ pin_prev,   blk ∉ pin       -> unpin to head
        elif blk in pin_prev and blk not in pin:
            pin[blk] = None     # XXX = head

        # blk ∈ pin_prev,   blk ∈ pin       -> if rev different: use pin
        elif blk in pin_prev and blk in pin:
            assert pin_prev[blk] <= pin[blk]
            if pin_prev[blk] == pin[blk]:
                del pin[blk]    # would need to pin to what it is already pinned

    #print('-> %s' % pinstr(pin))

    # {} blk -> at that have to be pinned
    # XXX also check that head/file[blk] is in cache - else no need to pin
    pinok = pin
    print('#  pinok: %s' % pinstr(pinok))

    # send watch request and check that we receive pins for in-cache blocks
    # changed > at.
    # XXX use timeout to detect wcfs replying less pins.
    ctx, cancel = context.with_cancel(context.background())
    wg = sync.WorkGroup(ctx)

    def _(ctx):
        pinv = w._expectPin(ctx, zf, pinok)
        for p in pinv:
            p.reply(b"ack")     # XXX -> return to caller?
        # check that we don't get extra pins before "ok" reply to "watch"
        try:
            req = w.recvReq(ctx)
        except Exception as e:
            if e is context.canceled:
                return # cancel is expected after seeing "ok"
            reraise(e, None, e.__traceback__)

        assert False, "extra pin message received: %r" % req.msg
    wg.go(_)

    def _(ctx):
        assert w.sendReq(ctx, b"watch %s @%s" % (h(zf._p_oid), h(at))) == "ok"
        # cancel _expectPin waiting upon receiving "ok" from wcfs
        # -> error that missed pins were not received.
        cancel()
    wg.go(_)

    wg.wait()
    w._watching[zf] = at

# _expectPin asserts that wcfs sends expected pin messages.
#
# expect is {} blk -> at
# returns [] of received pin requests.
@func(tWatch)
def _expectPin(w, ctx, zf, expect):
    expected = set()    # of expected pin messages
    for blk, at in expect.items():
        msg = b"pin %s #%d @%s" % (h(zf._p_oid), blk, h(at))
        assert msg not in expected
        expected.add(msg)

    reqv = []   # of received requests
    while len(expected) > 0:
        try:
            req = w.recvReq(ctx)
        except Exception as e:
            raise RuntimeError("%s\nnot all pin missages received - pending:\n%s" % (e, expected))
        assert req is not None  # channel not closed
        assert req.msg in expected
        expected.remove(req.msg)
        reqv.append(req)

    return reqv



# test_wcfs exercises wcfs functionality.
@func
def test_wcfs():
    t = tDB()
    defer(t.close)

    t.root['!file'] = nonfile = Persistent()
    t.root['zfile'] = zf = ZBigFile(blksize)

    at0  = t.commit()

    # >>> lookup non-BigFile -> must be rejected
    with raises(OSError) as exc:
        t.stat(nonfile)
    assert exc.value.errno == EINVAL

    # >>> file initially empty
    f = t.open(zf)
    f.assertCache([])
    f.assertData ([], mtime=at0)

    # >>> (@at1) commit data -> we can see it on wcfs
    t.change(zf, {2: b'alpha'})
    at1 = t.commit()

    f.assertCache([0,0,0])  # initially not cached
    f.assertData ([b'',b'',b'alpha'], mtime=t.head)

    # >>> (@at2) commit again -> we can see both latest and snapshotted states
    t.change(zf, {2: b'beta', 3: b'gamma'})
    at2 = t.commit()

    # f @head
    f.assertCache([1,1,0,0])
    f.assertData ([b'',b'', b'beta', b'gamma'], mtime=t.head)

    # f @at1
    f1 = t.open(zf, at=at1)
    f1.assertCache([0,0,1])
    f1.assertData ([b'',b'',b'alpha'])  # XXX + mtime=at1?


    # >>> (@at3) commit again without changing zf size
    f2 = t.open(zf, at=at2)
    t.change(zf, {2: b'kitty'})
    at3 = t.commit()

    f.assertCache([1,1,0,1])

    # f @head is opened again -> cache must not be lost
    f_ = t.open(zf)
    f_.assertCache([1,1,0,1])
    f_.close()
    f.assertCache([1,1,0,1])

    # f @head
    f.assertCache([1,1,0,1])
    f.assertData ([b'',b'',b'kitty',b'gamma'], mtime=t.head)

    # f @at2
    f2.assertCache([0,0,1,0])
    f2.assertData ([b'',b'',b'beta',b'gamma'])   # XXX mtime=at2

    # f @at1
    f1.assertCache([1,1,1])
    f1.assertData ([b'',b'',b'alpha'])  # XXX + mtime=at1?


    # >>> f close / open again -> cache must not be lost
    # XXX a bit flaky since OS can evict whole f cache under pressure
    f.assertCache([1,1,1,1])
    f.close()
    f = t.open(zf)
    assert f.cached() != [0,0,0,0]


    # >>> XXX commit data to not yet accessed f part - nothing happens

    # >>> invalidation protocol
    print('\n\n inv. protocol \n\n')

    for zf in t.zfiles():
        # watch from scratch
        for dF in t.dFtail:
            w = t.openwatch()
            w.watch(zf, dF.rev)
            w.close()

        # watch going at_i -> at_j
        for dF_from in t.dFtail:
            for dF_to in t.dFtail:
                if not (dF_from.rev <= dF_to.rev):
                    continue    # FIXME TODO test all directions

                print('\n--------')
                w = t.openwatch()
                w.watch(zf, dF_from.rev)
                w.watch(zf, dF_to.rev)
                w.close()

        # watch going at1 -> at2 -> at3
        print('\n--------')

    print()


    # XXX both from scratch and going e.g. at1 -> at2 -> at3
    # XXX going not only up, but also down at1 <- at2 <- at3    ?

    # XXX 2 (or more) opened watch for 1 file at the same time
    # XXX watch for 2 files via single watch open

    # XXX watch with @at > head - must wait for head to become >= at
    # XXX watch with @at < δtail.tail -> rejected
    # XXX watch with at="-" -> stop watching

    # XXX drop file[blk] from cache, access again -> no pin message sent the second time

    # XXX mmap f; change f[blk] on pin message while under pagefault - should get changed page

    # XXX access to block not previously accessed but invalidated in ZODB

    # XXX new watch request while previous watch request is in progress (over the same /head/watch handle)

    # XXX invalid requests -> wcfs replies with error

    # XXX blk not initially covered by f.δtail (blk never accessed - f.δtail
    # not updated on invalidation). then blk is accessed - what happens with
    # watch that should be triggerring for this blk?

    # XXX similar to ^^^ but with two changes to blk not covered by f.δtail. To
    # which @rev blk is pinned on watch? (δtail is missing both and if it will
    # be another blk rev < rev1,rev2 - it will be incorrect)

    # XXX pin message when blk data only first appeared after > w.at - pin
    # needs to pin to zero (or at0 ?).

    # XXX watch @at when file did not existed -> error

    # XXX ZBlk copied from blk1 -> blk2 ; for the same file and for file1 -> file2
    # XXX ZBlk moved  from blk1 -> blk2 ; for the same file and for file1 -> file2

    # XXX read file[blk]=hole; then file[blk]=zblk - must be invalidated and
    # setupWatch must send pins.


def test_wcfs_invproto():
    # XXX
    pass


# ---- misc ---

# readfile reads file @ path.
def readfile(path):
    with open(path) as f:
        return f.read()

# tidtime converts tid to transaction commit time.
def tidtime(tid):
    t = TimeStamp(tid).timeTime()

    # ZODB/py vs ZODB/go time resolution is not better than 1µs
    # see e.g. https://lab.nexedi.com/kirr/neo/commit/9112f21e
    #
    # NOTE pytest.approx supports only ==, not e.g. <, so we use plain round.
    return round(t, 6)

# verify that tidtime is precise enough to show difference in between transactions.
@func
def test_tidtime_notrough():
    t = tDB()
    defer(t.close)

    atprev = t.commit()
    for i in range(10):
        at = t.commit()
        assert tidtime(at) > tidtime(atprev)