Merge branch 't' into t2

* t: (129 commits)
  X go.mod: v↑ *
  go/zodb/btree: Change V<op> family to also provide visited node key coverage on visit callback
  go/zodb/btree: Add KeyRange type
  go/zodb/btree: Introduce constants for min/max key value
  go/zodb/btree: tests: Don't forget to close storage
  go/zodb/btree: Cosmetics
  .
  .
  .
  .
  .
  .
  .
  X Start reworking BTree to provide keycov on visit callback
  X go/neo: Fix credentials parsing with go1.17
  fixup! Y client: Fix URI scheme to move credentials out of query
  go/internal/xtesting: Add missing X
  go/zodb/{fs1,zeo}: ~staticcheck
  go/zodb/btree: Fix missing return on data-consistency error
  go/zodb, go/zodb/btree: Fix go generate after rename on zodbtools side
  ...

go/go.mod

@@ -3,24 +3,36 @@ module lab.nexedi.com/kirr/neo/go
 go 1.14
 
 require (
-	github.com/DataDog/czlib v0.0.0-20160811164712-4bc9a24e37f2
+	github.com/DataDog/czlib v0.0.0-20210322182103-8087f4e14ae7
 	github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548 // indirect
 	github.com/cznic/strutil v0.0.0-20181122101858-275e90344537
-	github.com/fsnotify/fsnotify v1.4.10-0.20200417215612-7f4cf4dd2b52
-	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
-	github.com/gwenn/gosqlite v0.0.0-20200521090053-24878be1a237
-	github.com/kisielk/og-rek v1.0.1-0.20180928202415-8b25c4cefd6c
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/fsnotify/fsnotify v1.5.1
+	github.com/golang/glog v1.0.0
+	github.com/golang/protobuf v1.4.3 // indirect
+	github.com/google/go-cmp v0.5.4 // indirect
+	github.com/gwenn/gosqlite v0.0.0-20211101095637-b18efb2e44c8
+	github.com/gwenn/yacr v0.0.0-20211101095056-492fb0c571bc // indirect
+	github.com/kisielk/og-rek v1.2.0
+	github.com/kr/text v0.2.0 // indirect
 	github.com/kylelemons/godebug v1.1.0
-	github.com/philhofer/fwd v1.0.0 // indirect
+	github.com/philhofer/fwd v1.1.1
 	github.com/pkg/errors v0.9.1
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
-	github.com/shamaton/msgpack v1.1.1
-	github.com/soheilhy/cmux v0.1.4
+	github.com/shamaton/msgpack v1.2.1
+	github.com/soheilhy/cmux v0.1.5
 	github.com/someonegg/gocontainer v1.0.0
-	github.com/stretchr/testify v1.6.1
-	github.com/tinylib/msgp v1.1.3-0.20200327023543-e88e92c0ccca
-	golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208
-	google.golang.org/grpc v1.32.0 // indirect
-	google.golang.org/grpc/examples v0.0.0-20200915000551-32e7099cccac // indirect
-	lab.nexedi.com/kirr/go123 v0.0.0-20200915142026-a281a51cf49b
+	github.com/stretchr/testify v1.7.0
+	github.com/tinylib/msgp v1.1.6
+	github.com/ttacon/chalk v0.0.0-20160626202418-22c06c80ed31 // indirect
+	golang.org/x/net v0.0.0-20211111160137-58aab5ef257a // indirect
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
+	golang.org/x/sys v0.0.0-20211111213525-f221eed1c01e // indirect
+	golang.org/x/text v0.3.7 // indirect
+	google.golang.org/genproto v0.0.0-20210226172003-ab064af71705 // indirect
+	google.golang.org/grpc v1.36.0 // indirect
+	google.golang.org/grpc/examples v0.0.0-20210301210255-fc8f38cccf75 // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+	lab.nexedi.com/kirr/go123 v0.0.0-20210906140734-c9eb28d9e408
 )

go/internal/xtesting/xtesting.go

-// Copyright (C) 2017-2020  Nexedi SA and Contributors.
+// Copyright (C) 2017-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -29,10 +29,13 @@ import (
 	"os"
 	"os/exec"
 	"reflect"
+	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"lab.nexedi.com/kirr/go123/xerr"
+	"lab.nexedi.com/kirr/go123/xstrings"
 	"lab.nexedi.com/kirr/neo/go/zodb"
 )
 
@@ -92,13 +95,14 @@ func NeedPy(t testing.TB, modules ...string) {
 }
 
 // ZRawObject represents raw ZODB object state.
-type ZRawObject struct {
+type ZRawObject struct { // keep in sync with zodb(test).ZRawObject
 	Oid  zodb.Oid
 	Data []byte // raw serialized zodb data
 }
 
 // ZPyCommitRaw commits new transaction into database @ zurl with raw data specified by objv.
 //
+// Nil data means "delete object".
 // The commit is performed via zodbtools/py.
 func ZPyCommitRaw(zurl string, at zodb.Tid, objv ...ZRawObject) (_ zodb.Tid, err error) {
 	defer xerr.Contextf(&err, "%s: zpycommit @%s", zurl, at)
@@ -109,9 +113,14 @@ func ZPyCommitRaw(zurl string, at zodb.Tid, objv ...ZRawObject) (_ zodb.Tid, err
 	fmt.Fprintf(zin, "description %q\n", fmt.Sprintf("test commit; at=%s", at))
 	fmt.Fprintf(zin, "extension %q\n", "")
 	for _, obj := range objv {
-		fmt.Fprintf(zin, "obj %s %d null:00\n", obj.Oid, len(obj.Data))
-		zin.Write(obj.Data)
-		zin.WriteString("\n")
+		// !data -> delete
+		if obj.Data == nil {
+			fmt.Fprintf(zin, "obj %s delete\n", obj.Oid)
+		} else {
+			fmt.Fprintf(zin, "obj %s %d null:00\n", obj.Oid, len(obj.Data))
+			zin.Write(obj.Data)
+			zin.WriteString("\n")
+		}
 	}
 	zin.WriteString("\n")
 
@@ -135,6 +144,33 @@ func ZPyCommitRaw(zurl string, at zodb.Tid, objv ...ZRawObject) (_ zodb.Tid, err
 
 // XXX + ZPyCommitSrv ?
 
+// ZPyRestore restores transactions specified by zin in zodbdump format.
+//
+// The restore is performed via zodbtools/py.
+func ZPyRestore(zurl string, zin string) (tidv []zodb.Tid, err error) {
+	defer xerr.Contextf(&err, "%s: zpyrestore", zurl)
+
+	// run py `zodb restore`
+	cmd:= exec.Command("python", "-m", "zodbtools.zodb", "restore", zurl)
+	cmd.Stdin  = strings.NewReader(zin)
+	cmd.Stderr = os.Stderr
+	out, err := cmd.Output()
+	if err != nil {
+		return nil, err
+	}
+
+	for _, line := range xstrings.SplitLines(string(out), "\n") {
+		tid, err := zodb.ParseTid(line)
+		if err != nil {
+			return nil, fmt.Errorf("restored, but invalid output: %s", err)
+		}
+
+		tidv = append(tidv, tid)
+	}
+
+	return tidv, nil
+}
+
 
 // ---- tests for storage drivers ----
 
@@ -352,7 +388,7 @@ func DrvTestWatch(t *testing.T, zurl string, zdrvOpen zodb.DriverOpener, bugv ..
 
 	ctx := context.Background()
 	watchq := make(chan zodb.Event)
-	zdrv, at0, err := zdrvOpen(ctx, u, &zodb.DriverOptions{ReadOnly: true, Watchq: watchq})
+	zdrv, at0, err := zdrvOpen(ctx, u, &zodb.DriverOptions{ReadOnly: true, Watchq: watchq}); X(err)
 	if at0 != at {
 		t.Fatalf("opened @ %s  ; want %s", at0, at)
 	}
@@ -438,6 +474,11 @@ func DrvTestWatch(t *testing.T, zurl string, zdrvOpen zodb.DriverOpener, bugv ..
 		}
 	}
 
+	// commit something more and wait a bit to raise chances the driver enqueues to watchq<- .
+	_ = xcommit(at, ZRawObject{0, b("at the end")})
+	time.Sleep(1*time.Second)
+
+	// the driver must handle Close and cancel that watchq<-
 	err = zdrv.Close(); X(err)
 
 	e, ok := <-watchq
@@ -450,6 +491,7 @@ func DrvTestWatch(t *testing.T, zurl string, zdrvOpen zodb.DriverOpener, bugv ..
 // FatalIf(t) returns function f(err), which call t.Fatal if err != nil.
 func FatalIf(t *testing.T) func(error) {
 	return func(err error) {
+		t.Helper()
 		if err != nil {
 			t.Fatal(err)
 		}

go/neo/client.go

@@ -68,6 +68,9 @@ type Client struct {
 	at0Initialized bool                // true after .at0 is initialized
 	at0Ready       chan(struct{})      // ready after .at0 is initialized
 
+	closeOnce sync.Once
+	closed    chan struct{} // ready when Closed
+
 	ownNet bool // true if Client "owns" networker and should release it on Close
 }
 
@@ -79,8 +82,9 @@ var _ zodb.IStorageDriver = (*Client)(nil)
 // Use Run to actually start running the node.
 func NewClient(clusterName, masterAddr string, net xnet.Networker) *Client {
 	c := &Client{
-		node:    newMasteredNode(proto.CLIENT, clusterName, net, masterAddr),
+		node:     newMasteredNode(proto.CLIENT, clusterName, net, masterAddr),
 		at0Ready: make(chan struct{}),
+		closed:   make(chan struct{}),
 	}
 
 	var runCtx context.Context
@@ -91,6 +95,9 @@ func NewClient(clusterName, masterAddr string, net xnet.Networker) *Client {
 
 // Close implements zodb.IStorageDriver.
 func (c *Client) Close() (err error) {
+	c.closeOnce.Do(func() {
+		close(c.closed)
+	})
 	c.runCancel()
 	err = c.runWG.Wait()
 	if errors.Is(err, context.Canceled) {
@@ -190,7 +197,7 @@ func (c *Client) invalidateObjects(msg *proto.InvalidateObjects) error {
 	defer c.at0Mu.Unlock()
 
 	// queue initial events until .at0 is initialized after register
-	// queued events will be sent to watchq by zeo ctor after initializing .at0
+	// queued events will be sent to watchq by syncMaster after initializing .at0
 	if !c.at0Initialized {
 		c.eventq0 = append(c.eventq0, event)
 		return nil
@@ -198,7 +205,13 @@ func (c *Client) invalidateObjects(msg *proto.InvalidateObjects) error {
 
 	// at0 is initialized - ok to send current event if it goes > at0
 	if tid > c.at0 {
-		c.watchq <- event
+		select {
+		case <-c.closed:
+			// closed - client does not read watchq anymore
+
+		case c.watchq <- event:
+			// ok
+		}
 	}
 	return nil
 }
@@ -250,7 +263,13 @@ func (c *Client) flushEventq0() {
 	if c.watchq != nil {
 		for _, e := range c.eventq0 {
 			if e.Tid > c.at0 {
-				c.watchq <- e
+				select {
+				case <-c.closed:
+					// closed - client does not read watchq anymore
+
+				case c.watchq <- e:
+					// ok
+				}
 			}
 		}
 	}
@@ -410,6 +429,8 @@ func openClientByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (
 	}
 
 	cred := u.User.String()
+	// ca=ca.crt;cert=my.crt;key=my.key
+	cred = strings.ReplaceAll(cred, ";", "&") // ; is no longer in default separators set https://github.com/golang/go/issues/25192
 	x, err := xurl.ParseQuery(cred)
 	if err != nil {
 		return nil, zodb.InvalidTid, fmt.Errorf("credentials: %s", err)
@@ -442,9 +463,7 @@ func openClientByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (
 	}
 
 	name := u.Path
-	if strings.HasPrefix(name, "/") {
-		name = name[1:]
-	}
+	name = strings.TrimPrefix(name, "/")
 	if name == "" {
 		return nil, zodb.InvalidTid, fmt.Errorf("cluster name not specified")
 	}
@@ -491,14 +510,18 @@ func openClientByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (
 		// close .watchq after serve is over
 		c.at0Mu.Lock()
 		defer c.at0Mu.Unlock()
-		if c.at0Initialized {
-			c.flushEventq0()
-		}
 		if c.watchq != nil {
-			if err != nil {
-				c.watchq <- &zodb.EventError{Err: err}
+			if err != nil && /* already flushed .eventq0 */c.at0Initialized {
+				select {
+				case <-c.closed:
+					// closed - client does not read watchq anymore
+
+				case c.watchq <- &zodb.EventError{Err: err}:
+					// ok
+				}
 			}
 			close(c.watchq)
+			c.watchq = nil // prevent flushEventq0 to send to closed chan
 		}
 
 		errq <- err

go/neo/client_test.go

@@ -362,6 +362,7 @@ func (opt NEOSrvOptions) URLPrefix() string {
 // ----------------
 
 // tOptions represents options for testing.
+// TODO -> xtesting
 type tOptions struct {
 	Preload string // preload database with data from this location
 }

go/neo/mastered.go

@@ -76,12 +76,12 @@ type _MasteredNode struct {
 type _MasteredNodeFlags int
 const (
 	// δPartTabPassThrough tells mlink.Recv1 not to filter out messages related
-	// to partition table changes. When mlink.Recv1 receives such messages there
+	// to partition table changes. When mlink.Recv1 receives such messages they
 	// are already processed internally to update .state.PartTab correspondingly.
 	//
 	// Storage uses this mode to receive δPartTab notifications to know
 	// when to persist it.
-	δPartTabPassThrough _MasteredNodeFlags = iota
+	δPartTabPassThrough _MasteredNodeFlags = 1 << iota
 )
 
 // newMasteredNode creates new _MasteredNode that connects to masterAddr/cluster via net.

go/neo/t/neotest

 #!/bin/bash -e
 # neotest: run tests and benchmarks against FileStorage, ZEO and various NEO/py and NEO/go clusters
 
-# Copyright (C) 2017-2020  Nexedi SA and Contributors.
+# Copyright (C) 2017-2021  Nexedi SA and Contributors.
 #                          Kirill Smelkov <kirr@nexedi.com>
 #
 # This program is free software: you can Use, Study, Modify and Redistribute
@@ -141,7 +141,10 @@ $@
 
 # ---- go/py unit tests ----
 cmd_test-go() {
-	go test -count=1 lab.nexedi.com/kirr/neo/go/... # -count=1 disables tests caching
+	(
+		cd $NEOt
+		go test -count=1 lab.nexedi.com/kirr/neo/go/... # -count=1 disables tests caching
+	)
 }
 
 cmd_test-py() {
@@ -1503,18 +1506,22 @@ cpustat)	f=(                     );;
 esac
 
 
+NEOt=$(cd `dirname $0` && pwd)
+
 for flag in ${f[*]}; do
 	case "$flag" in
 	build)
 		# make sure tzodb*, tcpu* and zgenprod are on PATH (because we could be invoked from another dir)
-		X=$(cd `dirname $0` && pwd)
-		export PATH=$X:$PATH
+		export PATH=$NEOt:$PATH
 
 		# rebuild go bits
 		# neo/py, wendelin.core, ... - must be pip install'ed - `neotest deploy` cares about that
-		go install -v lab.nexedi.com/kirr/neo/go/...
-		go build -o $X/tzodb_go $X/tzodb.go
-		go build -o $X/tcpu_go $X/tcpu.go
+		(
+			cd $NEOt
+			go install -v lab.nexedi.com/kirr/neo/go/...
+			go build -o tzodb_go tzodb.go
+			go build -o tcpu_go tcpu.go
+		)
 		;;
 
 	net)

go/neo/t/tzodb.go

@@ -431,7 +431,7 @@ func zwrkPreconnect(ctx context.Context, url string, at zodb.Tid, nwrk int) (_ [
 	if err != nil {
 		for _, stor := range storv {
 			if stor != nil {
-				xio.LClose(stor)
+				xio.LClose(ctx, stor)
 			}
 		}
 		return nil, err

go/zodb/btree/btree.go.in

 // Copyright (c) 2001, 2002 Zope Foundation and Contributors.
 // All Rights Reserved.
 //
-// Copyright (C) 2018-2019  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This software is subject to the provisions of the Zope Public License,
@@ -65,7 +65,7 @@ type BTree struct {
 
 // Entry is one BTree node entry.
 //
-// It contains key and child, who is either BTree or Bucket.
+// It contains key and child, which is either BTree or Bucket.
 //
 // Key limits child's keys - see BTree.Entryv for details.
 type Entry struct {
@@ -102,6 +102,15 @@ type BucketEntry struct {
 	value interface{}
 }
 
+// KeyRange represents [lo,hi) key range.
+type KeyRange struct {
+	Lo  KEY
+	Hi_ KEY // NOTE _not_ hi) to avoid overflow at ∞;  hi = hi_ + 1
+}
+
+const _KeyMin KEY = math.Min<Key>
+const _KeyMax KEY = math.Max<Key>
+
 // ---- access []entry ----
 
 // Key returns BTree entry key.
@@ -123,7 +132,7 @@ func (e *Entry) Child() Node { return e.child }
 // Children of all entries are guaranteed to be of the same kind - either all BTree, or all Bucket.
 //
 // The caller must not modify returned array.
-func (t *BTree) Entryv() []Entry {
+func (t *BTree) Entryv() /*readonly*/ []Entry {
 	return t.data
 }
 
@@ -134,7 +143,7 @@ func (e *BucketEntry) Key() KEY { return e.key }
 func (e *BucketEntry) Value() interface{} { return e.value }
 
 // Entryv returns entries of a Bucket node.
-func (b *Bucket) Entryv() []BucketEntry {
+func (b *Bucket) Entryv() /*readonly*/ []BucketEntry {
 	ev := make([]BucketEntry, len(b.keys))
 	for i, k := range b.keys {
 		ev[i] = BucketEntry{k, b.values[i]}
@@ -168,36 +177,54 @@ func (t *BTree) Get(ctx context.Context, key KEY) (_ interface{}, _ bool, err er
 // VGet is like Get but also calls visit while traversing the tree.
 //
 // Visit is called with node being activated.
-func (t *BTree) VGet(ctx context.Context, key KEY, visit func(node Node)) (_ interface{}, _ bool, err error) {
+func (t *BTree) VGet(ctx context.Context, key KEY, visit func(node Node, keycov KeyRange)) (_ interface{}, _ bool, err error) {
 	defer xerr.Contextf(&err, "btree(%s): get %v", t.POid(), key)
 	err = t.PActivate(ctx)
 	if err != nil {
 		return nil, false, err
 	}
 
+	keycov := KeyRange{Lo: _KeyMin, Hi_: _KeyMax}
 	if visit != nil {
-		visit(t)
-	}
-
-	if len(t.data) == 0 {
-		// empty btree
-		t.PDeactivate()
-		return nil, false, nil
+		visit(t, keycov)
 	}
 
 	for {
+		l := len(t.data)
+		if l == 0 {
+			// empty btree (top, or in leaf)
+			t.PDeactivate()
+			return nil, false, nil
+		}
+
 		// search i: K(i) ≤ k < K(i+1)	; K(0) = -∞, K(len) = +∞
-		i := sort.Search(len(t.data), func(i int) bool {
+		i := sort.Search(l, func(i int) bool {
 			j := i + 1
 			if j == len(t.data) {
 				return true // [len].key = +∞
 			}
 			return key < t.data[j].key
 		})
-
-		// FIXME panic index out of range (empty T without children;
-		// logically incorrect, but Restructure generated it once)
+		// i < l
 		child := t.data[i].child
+
+		// shorten global keycov by local [lo,hi) for this child
+		lo := _KeyMin
+		if i > 0 {
+			lo = t.data[i].key
+		}
+		i++
+		hi_ := _KeyMax
+		if i < l {
+			hi_ = t.data[i].key
+		}
+		if hi_ != _KeyMax {
+			hi_--
+		}
+		keycov.Lo  = kmax(keycov.Lo, lo)
+		keycov.Hi_ = kmin(keycov.Hi_, hi_)
+
+
 		t.PDeactivate()
 		err = child.PActivate(ctx)
 		if err != nil {
@@ -207,7 +234,7 @@ func (t *BTree) VGet(ctx context.Context, key KEY, visit func(node Node)) (_ int
 		// XXX verify child keys are in valid range according to parent
 
 		if visit != nil {
-			visit(child)
+			visit(child, keycov)
 		}
 
 		switch child := child.(type) {
@@ -250,26 +277,40 @@ func (t *BTree) MinKey(ctx context.Context) (_ KEY, ok bool, err error) {
 // VMinKey is like MinKey but also calls visit while traversing the tree.
 //
 // Visit is called with node being activated.
-func (t *BTree) VMinKey(ctx context.Context, visit func(node Node)) (_ KEY, ok bool, err error) {
+func (t *BTree) VMinKey(ctx context.Context, visit func(node Node, keycov KeyRange)) (_ KEY, ok bool, err error) {
 	defer xerr.Contextf(&err, "btree(%s): minkey", t.POid())
 	err = t.PActivate(ctx)
 	if err != nil {
 		return 0, false, err
 	}
 
+	keycov := KeyRange{Lo: _KeyMin, Hi_: _KeyMax}
 	if visit != nil {
-		visit(t)
-	}
-
-	if len(t.data) == 0 {
-		// empty btree
-		t.PDeactivate()
-		return 0, false, nil
+		visit(t, keycov)
 	}
 
 	// NOTE -> can also use t.firstbucket
 	for {
+		l := len(t.data)
+		if l == 0 {
+			// empty btree (top, or in leaf)
+			t.PDeactivate()
+			return 0, false, nil
+		}
+
 		child := t.data[0].child
+
+		// shorten global keycov by local hi) for this child
+		hi_ := _KeyMax
+		if 1 < l {
+			hi_ = t.data[1].key
+		}
+		if hi_ != _KeyMax {
+			hi_--
+		}
+		// keycov.Lo stays -∞
+		keycov.Hi_ = kmin(keycov.Hi_, hi_)
+
 		t.PDeactivate()
 		err = child.PActivate(ctx)
 		if err != nil {
@@ -277,7 +318,7 @@ func (t *BTree) VMinKey(ctx context.Context, visit func(node Node)) (_ KEY, ok b
 		}
 
 		if visit != nil {
-			visit(child)
+			visit(child, keycov)
 		}
 
 		// XXX verify child keys are in valid range according to parent
@@ -305,26 +346,36 @@ func (t *BTree) MaxKey(ctx context.Context) (_ KEY, _ bool, err error) {
 // VMaxKey is like MaxKey but also calls visit while traversing the tree.
 //
 // Visit is called with node being activated.
-func (t *BTree) VMaxKey(ctx context.Context, visit func(node Node)) (_ KEY, _ bool, err error) {
+func (t *BTree) VMaxKey(ctx context.Context, visit func(node Node, keycov KeyRange)) (_ KEY, _ bool, err error) {
 	defer xerr.Contextf(&err, "btree(%s): maxkey", t.POid())
 	err = t.PActivate(ctx)
 	if err != nil {
 		return 0, false, err
 	}
 
+	keycov := KeyRange{Lo: _KeyMin, Hi_: _KeyMax}
 	if visit != nil {
-		visit(t)
-	}
-
-	l := len(t.data)
-	if l == 0 {
-		// empty btree
-		t.PDeactivate()
-		return 0, false, nil
+		visit(t, keycov)
 	}
 
 	for {
+		l := len(t.data)
+		if l == 0 {
+			// empty btree (top, or in leaf)
+			t.PDeactivate()
+			return 0, false, nil
+		}
+
 		child := t.data[l-1].child
+
+		// shorten global keycov by local [lo for this chile
+		lo := _KeyMin
+		if l-1 > 0 {
+			lo = t.data[l-1].key
+		}
+		keycov.Lo  = kmax(keycov.Lo, lo)
+		// keycov.Hi_ stays ∞
+
 		t.PDeactivate()
 		err = child.PActivate(ctx)
 		if err != nil {
@@ -332,7 +383,7 @@ func (t *BTree) VMaxKey(ctx context.Context, visit func(node Node)) (_ KEY, _ bo
 		}
 
 		if visit != nil {
-			visit(child)
+			visit(child, keycov)
 		}
 
 		// XXX verify child keys are in valid range according to parent
@@ -591,7 +642,7 @@ func (bt *btreeState) PySetState(pystate interface{}) (err error) {
 	var kprev int64
 	var childrenKind int // 1 - BTree, 2 - Bucket
 	for i, idx := 0, 0; i < n; i++ {
-		key := int64(math.Min<Key>) // KEY(-∞)   (qualifies for ≤)
+		key := int64(_KeyMin) // KEY(-∞)   (qualifies for ≤)
 		if i > 0 {
 			// key[0] is unused and not saved
 			key, ok = t[idx].(int64) // XXX Xint
@@ -609,7 +660,7 @@ func (bt *btreeState) PySetState(pystate interface{}) (err error) {
 		}
 
 		if i > 1 && !(key > kprev) {
-			fmt.Errorf("data: [%d]: key not ↑", i)
+			return fmt.Errorf("data: [%d]: key not ↑", i)
 		}
 		kprev = key
 
@@ -629,7 +680,7 @@ func (bt *btreeState) PySetState(pystate interface{}) (err error) {
 			childrenKind = kind
 		}
 		if kind != childrenKind {
-			fmt.Errorf("data: [%d]: children must be of the same type", i)
+			return fmt.Errorf("data: [%d]: children must be of the same type", i)
 		}
 
 		bt.data = append(bt.data, Entry{key: kkey, child: child.(Node)})
@@ -646,3 +697,60 @@ func init() {
 	zodb.RegisterClass("BTrees.BTree.BTree",  t(BTree{}),  t(btreeState{}))
 	zodb.RegisterClass("BTrees.BTree.Bucket", t(Bucket{}), t(bucketState{}))
 }
+
+
+// ---- misc ----
+
+// Has returns whether key k belongs to the range.
+func (r *KeyRange) Has(k KEY) bool {
+	return (r.Lo <= k && k <= r.Hi_)
+}
+
+// Empty returns whether key range is empty.
+func (r *KeyRange) Empty() bool {
+	hi := r.Hi_
+	if hi == _KeyMax {
+		// [x,∞] cannot be empty because max x is ∞ and [∞,∞] has one element: ∞
+		return false
+	}
+	hi++ // no overflow
+	return r.Lo >= hi
+}
+
+func (r KeyRange) String() string {
+	var shi string
+	if r.Hi_ == _KeyMax {
+		shi = kstr(r.Hi_) // ∞
+	} else {
+		shi = fmt.Sprintf("%d", r.Hi_+1)
+	}
+	return fmt.Sprintf("[%s,%s)", kstr(r.Lo), shi)
+}
+
+
+func kmin(a, b KEY) KEY {
+	if a < b {
+		return a
+	} else {
+		return b
+	}
+}
+
+func kmax(a, b KEY) KEY {
+	if a > b {
+		return a
+	} else {
+		return b
+	}
+}
+
+// kstr formats key as string.
+func kstr(k KEY) string {
+	if k == _KeyMin {
+		return "-∞"
+	}
+	if k == _KeyMax {
+		return "∞"
+	}
+	return fmt.Sprintf("%d", k)
+}

go/zodb/btree/btree_test.go

-// Copyright (C) 2018-2019  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -23,6 +23,7 @@ package btree
 
 import (
 	"context"
+	"reflect"
 	"testing"
 
 	"lab.nexedi.com/kirr/go123/exc"
@@ -98,6 +99,12 @@ func (b *bucketWrap) MaxKey(ctx context.Context) (k int64, ok bool, err error) {
 	return
 }
 
+// tVisit is information about one visit call.
+type tVisit struct {
+	node   zodb.Oid
+	keycov LKeyRange
+}
+
 func TestBTree(t *testing.T) {
 	X := exc.Raiseif
 	ctx := context.Background()
@@ -105,7 +112,11 @@ func TestBTree(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	db := zodb.NewDB(stor)
+	defer func() {
+		err := stor.Close(); X(err)
+	}()
+
+	db := zodb.NewDB(stor, &zodb.DBOptions{})
 	defer func() {
 		err := db.Close(); X(err)
 	}()
@@ -118,8 +129,6 @@ func TestBTree(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// XXX close db/stor
-
 	// go through small test Buckets/BTrees and verify that Get(key) is as expected.
 	for _, tt := range smallTestv {
 		xobj, err := conn.Get(ctx, tt.oid)
@@ -262,4 +271,39 @@ func TestBTree(t *testing.T) {
 	// XXX verify FirstBucket / Next ?
 
 	verifyFirstBucket(B3)
+
+	// verify nodes/keycov visited through VGet/V{Min,Max}Key
+	xBv, err := conn.Get(ctx, Bv_oid);  X(err)
+	Bv, ok := xBv.(*LOBTree)
+	if !ok {
+		t.Fatalf("Bv: %v; got %T;  want LOBTree", Bv_oid, xBv)
+	}
+
+	for k, visitOK := range Bvdict {
+		visit := []tVisit{}
+		_, _, err := Bv.VGet(ctx, k, func(node LONode, keycov LKeyRange) {
+			visit = append(visit, tVisit{node.POid(), keycov})
+		}); X(err)
+		if !reflect.DeepEqual(visit, visitOK) {
+			t.Errorf("VGet(%d): visit:\nhave: %v\nwant: %v", k, visit, visitOK)
+		}
+	}
+
+	visitMinOK := Bvdict[Bv_kmin]
+	visitMaxOK := Bvdict[Bv_kmax]
+	visitMin := []tVisit{}
+	visitMax := []tVisit{}
+	_, _, err = Bv.VMinKey(ctx, func(node LONode, keycov LKeyRange) {
+		visitMin = append(visitMin, tVisit{node.POid(), keycov})
+	}); X(err)
+	_, _, err = Bv.VMaxKey(ctx, func(node LONode, keycov LKeyRange) {
+		visitMax = append(visitMax, tVisit{node.POid(), keycov})
+	}); X(err)
+
+	if !reflect.DeepEqual(visitMin, visitMinOK) {
+		t.Errorf("VMinKey(): visit:\nhave: %v\nwant: %v", visitMin, visitMinOK)
+	}
+	if !reflect.DeepEqual(visitMax, visitMaxOK) {
+		t.Errorf("VMaxKey(): visit:\nhave: %v\nwant: %v", visitMax, visitMaxOK)
+	}
 }

go/zodb/btree/gen-btree

@@ -28,6 +28,9 @@ out=$3
 kind=${KIND,,}	# IO -> io
 Key=${KEY^}
 
+KEYKIND=${KIND:0:1}	# IO -> I
+keykind=${KEYKIND,,}	# I -> i
+
 input=$(dirname $0)/btree.go.in
 
 echo "// Code generated by gen-btree; DO NOT EDIT." >$out
@@ -45,4 +48,10 @@ sed	\
 	-e "s/\bBucketEntry\b/${KIND}BucketEntry/g"	\
 	-e "s/\bbtreeState\b/${kind}btreeState/g"	\
 	-e "s/\bbucketState\b/${kind}bucketState/g"	\
+	-e "s/\b_KeyMin\b/_${KEYKIND}KeyMin/g"		\
+	-e "s/\b_KeyMax\b/_${KEYKIND}KeyMax/g"		\
+	-e "s/\bKeyRange\b/${KEYKIND}KeyRange/g"	\
+	-e "s/\bkmin\b/${keykind}kmin/g"		\
+	-e "s/\bkmax\b/${keykind}kmax/g"		\
+	-e "s/\bkstr\b/${keykind}kstr/g"		\
 	$input >>$out

go/zodb/btree/testdata/gen-testdata

 #!/usr/bin/env python2
 # -*- coding: utf-8 -*-
-# Copyright (C) 2018-2020  Nexedi SA and Contributors.
+# Copyright (C) 2018-2021  Nexedi SA and Contributors.
 #                          Kirill Smelkov <kirr@nexedi.com>
 #
 # This program is free software: you can Use, Study, Modify and Redistribute
@@ -22,8 +22,9 @@
 
 from ZODB.DB import DB
 from BTrees.LOBTree import LOBucket, LOBTree
+from BTrees.check import check as bcheck
 from ZODB.utils import u64
-from zodbtools.test.gen_testdata import run_with_zodb3py2_compat
+from zodbtools.test.gen_testdata import run_with_zodb4py2_compat
 import os, os.path, transaction
 
 from golang.gcompat import qq
@@ -52,14 +53,37 @@ def main2():
 
     root['B3'] = B3 = LOBTree(dict([(_, _) for _ in range(10000)]))
 
+
+    # T4/T2-T/B1-B2-T7,9/B5-B8-B10  (to verify VGet->visit)
+    # TODO use xbtree.py:Restructure after gimport works through modules and
+    #      xbtree.py is moved from wcfs to zodb/go.
+    v1 = LOBucket([(1,"a")])
+    v2 = LOBucket([(2,"b")])
+    v5 = LOBucket([(5,"c")])
+    v8 = LOBucket([(8,"d")])
+    v9 = LOBucket([(9,"e")])
+    T2, T79, T, T4 = LOBTree(), LOBTree(), LOBTree(), LOBTree()
+    T2.__setstate__ (((v1, 2, v2),         v1)) # (child, key, child, ...), firstbucket
+    T79.__setstate__(((v5, 7, v8, 9, v9),  v5))
+    T.__setstate__  (((T79,),              v5))
+    T4.__setstate__ (((T2, 4, T),          v1))
+    root['Bv'] = Bv = T4
+
     transaction.commit()
 
+    bcheck(Bv)
+    assert Bv[1] == "a"
+    assert Bv[2] == "b"
+    assert Bv[5] == "c"
+    assert Bv[8] == "d"
+    assert Bv[9] == "e"
+
+
     with open("ztestdata_expect_test.go", "w") as f:
         def emit(v):
             print >>f, v
         emit("// Code generated by %s; DO NOT EDIT." % __file__)
         emit("package btree\n")
-        #emit("import \"lab.nexedi.com/kirr/neo/go/zodb\"\n")
 
         def emititems(b):
             s = "testEntry{oid: %s, kind: %s, itemv: []kv{" \
@@ -84,12 +108,34 @@ def main2():
         emit("\nconst B3_oid = %s" % u64(B3._p_oid))
         emit("const B3_maxkey = %d" % B3.maxKey())
 
+        emit("\nconst Bv_oid = %s" % u64(Bv._p_oid))
+        emit("const Bv_kmin = %d"  % Bv.minKey())
+        emit("const Bv_kmax = %d"  % Bv.maxKey())
+        emit("var Bvdict = map[int64][]tVisit{")
+        noo = "_LKeyMin"
+        oo  = "_LKeyMax"
+        def emitVisit(key, *visitv): # visitv = [](node, lo,hi)
+            vstr = []
+            for node, lo,hi in visitv:
+                if isinstance(hi, str):
+                    hi_ = hi    # oo or noo
+                else:
+                    hi_ = hi-1
+                vstr.append("{%d, LKeyRange{%s, %s}}" % (u64(node._p_oid), lo, hi_))
+            emit("\t%d: []tVisit{%s}," % (key, ", ".join(vstr)))
+
+        emitVisit(1, (T4, noo,oo), (T2, noo,4), (v1, noo,2))
+        emitVisit(2, (T4, noo,oo), (T2, noo,4), (v2, 2,4))
+        emitVisit(5, (T4, noo,oo), (T,  4,oo),  (T79, 4,oo), (v5, 4,7))
+        emitVisit(8, (T4, noo,oo), (T,  4,oo),  (T79, 4,oo), (v8, 7,9))
+        emitVisit(9, (T4, noo,oo), (T,  4,oo),  (T79, 4,oo), (v9, 9,oo))
+        emit("}")
 
     conn.close()
     db.close()
 
 def main():
-    run_with_zodb3py2_compat(main2)
+    run_with_zodb4py2_compat(main2)
 
 
 if __name__ == '__main__':

go/zodb/btree/ziobtree.go

@@ -3,7 +3,7 @@
 // Copyright (c) 2001, 2002 Zope Foundation and Contributors.
 // All Rights Reserved.
 //
-// Copyright (C) 2018-2019  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This software is subject to the provisions of the Zope Public License,
@@ -67,7 +67,7 @@ type IOBTree struct {
 
 // IOEntry is one IOBTree node entry.
 //
-// It contains key and child, who is either IOBTree or IOBucket.
+// It contains key and child, which is either IOBTree or IOBucket.
 //
 // Key limits child's keys - see IOBTree.Entryv for details.
 type IOEntry struct {
@@ -104,6 +104,15 @@ type IOBucketEntry struct {
 	value interface{}
 }
 
+// IKeyRange represents [lo,hi) key range.
+type IKeyRange struct {
+	Lo  int32
+	Hi_ int32 // NOTE _not_ hi) to avoid overflow at ∞;  hi = hi_ + 1
+}
+
+const _IKeyMin int32 = math.MinInt32
+const _IKeyMax int32 = math.MaxInt32
+
 // ---- access []entry ----
 
 // Key returns IOBTree entry key.
@@ -125,7 +134,7 @@ func (e *IOEntry) Child() IONode { return e.child }
 // Children of all entries are guaranteed to be of the same kind - either all IOBTree, or all IOBucket.
 //
 // The caller must not modify returned array.
-func (t *IOBTree) Entryv() []IOEntry {
+func (t *IOBTree) Entryv() /*readonly*/ []IOEntry {
 	return t.data
 }
 
@@ -136,7 +145,7 @@ func (e *IOBucketEntry) Key() int32 { return e.key }
 func (e *IOBucketEntry) Value() interface{} { return e.value }
 
 // Entryv returns entries of a IOBucket node.
-func (b *IOBucket) Entryv() []IOBucketEntry {
+func (b *IOBucket) Entryv() /*readonly*/ []IOBucketEntry {
 	ev := make([]IOBucketEntry, len(b.keys))
 	for i, k := range b.keys {
 		ev[i] = IOBucketEntry{k, b.values[i]}
@@ -170,34 +179,54 @@ func (t *IOBTree) Get(ctx context.Context, key int32) (_ interface{}, _ bool, er
 // VGet is like Get but also calls visit while traversing the tree.
 //
 // Visit is called with node being activated.
-func (t *IOBTree) VGet(ctx context.Context, key int32, visit func(node IONode)) (_ interface{}, _ bool, err error) {
+func (t *IOBTree) VGet(ctx context.Context, key int32, visit func(node IONode, keycov IKeyRange)) (_ interface{}, _ bool, err error) {
 	defer xerr.Contextf(&err, "btree(%s): get %v", t.POid(), key)
 	err = t.PActivate(ctx)
 	if err != nil {
 		return nil, false, err
 	}
 
+	keycov := IKeyRange{Lo: _IKeyMin, Hi_: _IKeyMax}
 	if visit != nil {
-		visit(t)
-	}
-
-	if len(t.data) == 0 {
-		// empty btree
-		t.PDeactivate()
-		return nil, false, nil
+		visit(t, keycov)
 	}
 
 	for {
+		l := len(t.data)
+		if l == 0 {
+			// empty btree (top, or in leaf)
+			t.PDeactivate()
+			return nil, false, nil
+		}
+
 		// search i: K(i) ≤ k < K(i+1)	; K(0) = -∞, K(len) = +∞
-		i := sort.Search(len(t.data), func(i int) bool {
+		i := sort.Search(l, func(i int) bool {
 			j := i + 1
 			if j == len(t.data) {
 				return true // [len].key = +∞
 			}
 			return key < t.data[j].key
 		})
-
+		// i < l
 		child := t.data[i].child
+
+		// shorten global keycov by local [lo,hi) for this child
+		lo := _IKeyMin
+		if i > 0 {
+			lo = t.data[i].key
+		}
+		i++
+		hi_ := _IKeyMax
+		if i < l {
+			hi_ = t.data[i].key
+		}
+		if hi_ != _IKeyMax {
+			hi_--
+		}
+		keycov.Lo  = ikmax(keycov.Lo, lo)
+		keycov.Hi_ = ikmin(keycov.Hi_, hi_)
+
+
 		t.PDeactivate()
 		err = child.PActivate(ctx)
 		if err != nil {
@@ -207,7 +236,7 @@ func (t *IOBTree) VGet(ctx context.Context, key int32, visit func(node IONode))
 		// XXX verify child keys are in valid range according to parent
 
 		if visit != nil {
-			visit(child)
+			visit(child, keycov)
 		}
 
 		switch child := child.(type) {
@@ -250,26 +279,40 @@ func (t *IOBTree) MinKey(ctx context.Context) (_ int32, ok bool, err error) {
 // VMinKey is like MinKey but also calls visit while traversing the tree.
 //
 // Visit is called with node being activated.
-func (t *IOBTree) VMinKey(ctx context.Context, visit func(node IONode)) (_ int32, ok bool, err error) {
+func (t *IOBTree) VMinKey(ctx context.Context, visit func(node IONode, keycov IKeyRange)) (_ int32, ok bool, err error) {
 	defer xerr.Contextf(&err, "btree(%s): minkey", t.POid())
 	err = t.PActivate(ctx)
 	if err != nil {
 		return 0, false, err
 	}
 
+	keycov := IKeyRange{Lo: _IKeyMin, Hi_: _IKeyMax}
 	if visit != nil {
-		visit(t)
-	}
-
-	if len(t.data) == 0 {
-		// empty btree
-		t.PDeactivate()
-		return 0, false, nil
+		visit(t, keycov)
 	}
 
 	// NOTE -> can also use t.firstbucket
 	for {
+		l := len(t.data)
+		if l == 0 {
+			// empty btree (top, or in leaf)
+			t.PDeactivate()
+			return 0, false, nil
+		}
+
 		child := t.data[0].child
+
+		// shorten global keycov by local hi) for this child
+		hi_ := _IKeyMax
+		if 1 < l {
+			hi_ = t.data[1].key
+		}
+		if hi_ != _IKeyMax {
+			hi_--
+		}
+		// keycov.Lo stays -∞
+		keycov.Hi_ = ikmin(keycov.Hi_, hi_)
+
 		t.PDeactivate()
 		err = child.PActivate(ctx)
 		if err != nil {
@@ -277,7 +320,7 @@ func (t *IOBTree) VMinKey(ctx context.Context, visit func(node IONode)) (_ int32
 		}
 
 		if visit != nil {
-			visit(child)
+			visit(child, keycov)
 		}
 
 		// XXX verify child keys are in valid range according to parent
@@ -305,26 +348,36 @@ func (t *IOBTree) MaxKey(ctx context.Context) (_ int32, _ bool, err error) {
 // VMaxKey is like MaxKey but also calls visit while traversing the tree.
 //
 // Visit is called with node being activated.
-func (t *IOBTree) VMaxKey(ctx context.Context, visit func(node IONode)) (_ int32, _ bool, err error) {
+func (t *IOBTree) VMaxKey(ctx context.Context, visit func(node IONode, keycov IKeyRange)) (_ int32, _ bool, err error) {
 	defer xerr.Contextf(&err, "btree(%s): maxkey", t.POid())
 	err = t.PActivate(ctx)
 	if err != nil {
 		return 0, false, err
 	}
 
+	keycov := IKeyRange{Lo: _IKeyMin, Hi_: _IKeyMax}
 	if visit != nil {
-		visit(t)
-	}
-
-	l := len(t.data)
-	if l == 0 {
-		// empty btree
-		t.PDeactivate()
-		return 0, false, nil
+		visit(t, keycov)
 	}
 
 	for {
+		l := len(t.data)
+		if l == 0 {
+			// empty btree (top, or in leaf)
+			t.PDeactivate()
+			return 0, false, nil
+		}
+
 		child := t.data[l-1].child
+
+		// shorten global keycov by local [lo for this chile
+		lo := _IKeyMin
+		if l-1 > 0 {
+			lo = t.data[l-1].key
+		}
+		keycov.Lo  = ikmax(keycov.Lo, lo)
+		// keycov.Hi_ stays ∞
+
 		t.PDeactivate()
 		err = child.PActivate(ctx)
 		if err != nil {
@@ -332,7 +385,7 @@ func (t *IOBTree) VMaxKey(ctx context.Context, visit func(node IONode)) (_ int32
 		}
 
 		if visit != nil {
-			visit(child)
+			visit(child, keycov)
 		}
 
 		// XXX verify child keys are in valid range according to parent
@@ -591,7 +644,7 @@ func (bt *iobtreeState) PySetState(pystate interface{}) (err error) {
 	var kprev int64
 	var childrenKind int // 1 - IOBTree, 2 - IOBucket
 	for i, idx := 0, 0; i < n; i++ {
-		key := int64(math.MinInt32) // int32(-∞)   (qualifies for ≤)
+		key := int64(_IKeyMin) // int32(-∞)   (qualifies for ≤)
 		if i > 0 {
 			// key[0] is unused and not saved
 			key, ok = t[idx].(int64) // XXX Xint
@@ -609,7 +662,7 @@ func (bt *iobtreeState) PySetState(pystate interface{}) (err error) {
 		}
 
 		if i > 1 && !(key > kprev) {
-			fmt.Errorf("data: [%d]: key not ↑", i)
+			return fmt.Errorf("data: [%d]: key not ↑", i)
 		}
 		kprev = key
 
@@ -629,7 +682,7 @@ func (bt *iobtreeState) PySetState(pystate interface{}) (err error) {
 			childrenKind = kind
 		}
 		if kind != childrenKind {
-			fmt.Errorf("data: [%d]: children must be of the same type", i)
+			return fmt.Errorf("data: [%d]: children must be of the same type", i)
 		}
 
 		bt.data = append(bt.data, IOEntry{key: kkey, child: child.(IONode)})
@@ -646,3 +699,60 @@ func init() {
 	zodb.RegisterClass("BTrees.IOBTree.IOBTree",  t(IOBTree{}),  t(iobtreeState{}))
 	zodb.RegisterClass("BTrees.IOBTree.IOBucket", t(IOBucket{}), t(iobucketState{}))
 }
+
+
+// ---- misc ----
+
+// Has returns whether key k belongs to the range.
+func (r *IKeyRange) Has(k int32) bool {
+	return (r.Lo <= k && k <= r.Hi_)
+}
+
+// Empty returns whether key range is empty.
+func (r *IKeyRange) Empty() bool {
+	hi := r.Hi_
+	if hi == _IKeyMax {
+		// [x,∞] cannot be empty because max x is ∞ and [∞,∞] has one element: ∞
+		return false
+	}
+	hi++ // no overflow
+	return r.Lo >= hi
+}
+
+func (r IKeyRange) String() string {
+	var shi string
+	if r.Hi_ == _IKeyMax {
+		shi = ikstr(r.Hi_) // ∞
+	} else {
+		shi = fmt.Sprintf("%d", r.Hi_+1)
+	}
+	return fmt.Sprintf("[%s,%s)", ikstr(r.Lo), shi)
+}
+
+
+func ikmin(a, b int32) int32 {
+	if a < b {
+		return a
+	} else {
+		return b
+	}
+}
+
+func ikmax(a, b int32) int32 {
+	if a > b {
+		return a
+	} else {
+		return b
+	}
+}
+
+// ikstr formats key as string.
+func ikstr(k int32) string {
+	if k == _IKeyMin {
+		return "-∞"
+	}
+	if k == _IKeyMax {
+		return "∞"
+	}
+	return fmt.Sprintf("%d", k)
+}

go/zodb/btree/zlobtree.go

@@ -3,7 +3,7 @@
 // Copyright (c) 2001, 2002 Zope Foundation and Contributors.
 // All Rights Reserved.
 //
-// Copyright (C) 2018-2019  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This software is subject to the provisions of the Zope Public License,
@@ -67,7 +67,7 @@ type LOBTree struct {
 
 // LOEntry is one LOBTree node entry.
 //
-// It contains key and child, who is either LOBTree or LOBucket.
+// It contains key and child, which is either LOBTree or LOBucket.
 //
 // Key limits child's keys - see LOBTree.Entryv for details.
 type LOEntry struct {
@@ -104,6 +104,15 @@ type LOBucketEntry struct {
 	value interface{}
 }
 
+// LKeyRange represents [lo,hi) key range.
+type LKeyRange struct {
+	Lo  int64
+	Hi_ int64 // NOTE _not_ hi) to avoid overflow at ∞;  hi = hi_ + 1
+}
+
+const _LKeyMin int64 = math.MinInt64
+const _LKeyMax int64 = math.MaxInt64
+
 // ---- access []entry ----
 
 // Key returns LOBTree entry key.
@@ -125,7 +134,7 @@ func (e *LOEntry) Child() LONode { return e.child }
 // Children of all entries are guaranteed to be of the same kind - either all LOBTree, or all LOBucket.
 //
 // The caller must not modify returned array.
-func (t *LOBTree) Entryv() []LOEntry {
+func (t *LOBTree) Entryv() /*readonly*/ []LOEntry {
 	return t.data
 }
 
@@ -136,7 +145,7 @@ func (e *LOBucketEntry) Key() int64 { return e.key }
 func (e *LOBucketEntry) Value() interface{} { return e.value }
 
 // Entryv returns entries of a LOBucket node.
-func (b *LOBucket) Entryv() []LOBucketEntry {
+func (b *LOBucket) Entryv() /*readonly*/ []LOBucketEntry {
 	ev := make([]LOBucketEntry, len(b.keys))
 	for i, k := range b.keys {
 		ev[i] = LOBucketEntry{k, b.values[i]}
@@ -170,36 +179,54 @@ func (t *LOBTree) Get(ctx context.Context, key int64) (_ interface{}, _ bool, er
 // VGet is like Get but also calls visit while traversing the tree.
 //
 // Visit is called with node being activated.
-func (t *LOBTree) VGet(ctx context.Context, key int64, visit func(node LONode)) (_ interface{}, _ bool, err error) {
+func (t *LOBTree) VGet(ctx context.Context, key int64, visit func(node LONode, keycov LKeyRange)) (_ interface{}, _ bool, err error) {
 	defer xerr.Contextf(&err, "btree(%s): get %v", t.POid(), key)
 	err = t.PActivate(ctx)
 	if err != nil {
 		return nil, false, err
 	}
 
+	keycov := LKeyRange{Lo: _LKeyMin, Hi_: _LKeyMax}
 	if visit != nil {
-		visit(t)
-	}
-
-	if len(t.data) == 0 {
-		// empty btree
-		t.PDeactivate()
-		return nil, false, nil
+		visit(t, keycov)
 	}
 
 	for {
+		l := len(t.data)
+		if l == 0 {
+			// empty btree (top, or in leaf)
+			t.PDeactivate()
+			return nil, false, nil
+		}
+
 		// search i: K(i) ≤ k < K(i+1)	; K(0) = -∞, K(len) = +∞
-		i := sort.Search(len(t.data), func(i int) bool {
+		i := sort.Search(l, func(i int) bool {
 			j := i + 1
 			if j == len(t.data) {
 				return true // [len].key = +∞
 			}
 			return key < t.data[j].key
 		})
-
-		// FIXME panic index out of range (empty T without children;
-		// logically incorrect, but Restructure generated it once)
+		// i < l
 		child := t.data[i].child
+
+		// shorten global keycov by local [lo,hi) for this child
+		lo := _LKeyMin
+		if i > 0 {
+			lo = t.data[i].key
+		}
+		i++
+		hi_ := _LKeyMax
+		if i < l {
+			hi_ = t.data[i].key
+		}
+		if hi_ != _LKeyMax {
+			hi_--
+		}
+		keycov.Lo  = lkmax(keycov.Lo, lo)
+		keycov.Hi_ = lkmin(keycov.Hi_, hi_)
+
+
 		t.PDeactivate()
 		err = child.PActivate(ctx)
 		if err != nil {
@@ -209,7 +236,7 @@ func (t *LOBTree) VGet(ctx context.Context, key int64, visit func(node LONode))
 		// XXX verify child keys are in valid range according to parent
 
 		if visit != nil {
-			visit(child)
+			visit(child, keycov)
 		}
 
 		switch child := child.(type) {
@@ -252,26 +279,40 @@ func (t *LOBTree) MinKey(ctx context.Context) (_ int64, ok bool, err error) {
 // VMinKey is like MinKey but also calls visit while traversing the tree.
 //
 // Visit is called with node being activated.
-func (t *LOBTree) VMinKey(ctx context.Context, visit func(node LONode)) (_ int64, ok bool, err error) {
+func (t *LOBTree) VMinKey(ctx context.Context, visit func(node LONode, keycov LKeyRange)) (_ int64, ok bool, err error) {
 	defer xerr.Contextf(&err, "btree(%s): minkey", t.POid())
 	err = t.PActivate(ctx)
 	if err != nil {
 		return 0, false, err
 	}
 
+	keycov := LKeyRange{Lo: _LKeyMin, Hi_: _LKeyMax}
 	if visit != nil {
-		visit(t)
-	}
-
-	if len(t.data) == 0 {
-		// empty btree
-		t.PDeactivate()
-		return 0, false, nil
+		visit(t, keycov)
 	}
 
 	// NOTE -> can also use t.firstbucket
 	for {
+		l := len(t.data)
+		if l == 0 {
+			// empty btree (top, or in leaf)
+			t.PDeactivate()
+			return 0, false, nil
+		}
+
 		child := t.data[0].child
+
+		// shorten global keycov by local hi) for this child
+		hi_ := _LKeyMax
+		if 1 < l {
+			hi_ = t.data[1].key
+		}
+		if hi_ != _LKeyMax {
+			hi_--
+		}
+		// keycov.Lo stays -∞
+		keycov.Hi_ = lkmin(keycov.Hi_, hi_)
+
 		t.PDeactivate()
 		err = child.PActivate(ctx)
 		if err != nil {
@@ -279,7 +320,7 @@ func (t *LOBTree) VMinKey(ctx context.Context, visit func(node LONode)) (_ int64
 		}
 
 		if visit != nil {
-			visit(child)
+			visit(child, keycov)
 		}
 
 		// XXX verify child keys are in valid range according to parent
@@ -307,26 +348,36 @@ func (t *LOBTree) MaxKey(ctx context.Context) (_ int64, _ bool, err error) {
 // VMaxKey is like MaxKey but also calls visit while traversing the tree.
 //
 // Visit is called with node being activated.
-func (t *LOBTree) VMaxKey(ctx context.Context, visit func(node LONode)) (_ int64, _ bool, err error) {
+func (t *LOBTree) VMaxKey(ctx context.Context, visit func(node LONode, keycov LKeyRange)) (_ int64, _ bool, err error) {
 	defer xerr.Contextf(&err, "btree(%s): maxkey", t.POid())
 	err = t.PActivate(ctx)
 	if err != nil {
 		return 0, false, err
 	}
 
+	keycov := LKeyRange{Lo: _LKeyMin, Hi_: _LKeyMax}
 	if visit != nil {
-		visit(t)
-	}
-
-	l := len(t.data)
-	if l == 0 {
-		// empty btree
-		t.PDeactivate()
-		return 0, false, nil
+		visit(t, keycov)
 	}
 
 	for {
+		l := len(t.data)
+		if l == 0 {
+			// empty btree (top, or in leaf)
+			t.PDeactivate()
+			return 0, false, nil
+		}
+
 		child := t.data[l-1].child
+
+		// shorten global keycov by local [lo for this chile
+		lo := _LKeyMin
+		if l-1 > 0 {
+			lo = t.data[l-1].key
+		}
+		keycov.Lo  = lkmax(keycov.Lo, lo)
+		// keycov.Hi_ stays ∞
+
 		t.PDeactivate()
 		err = child.PActivate(ctx)
 		if err != nil {
@@ -334,7 +385,7 @@ func (t *LOBTree) VMaxKey(ctx context.Context, visit func(node LONode)) (_ int64
 		}
 
 		if visit != nil {
-			visit(child)
+			visit(child, keycov)
 		}
 
 		// XXX verify child keys are in valid range according to parent
@@ -593,7 +644,7 @@ func (bt *lobtreeState) PySetState(pystate interface{}) (err error) {
 	var kprev int64
 	var childrenKind int // 1 - LOBTree, 2 - LOBucket
 	for i, idx := 0, 0; i < n; i++ {
-		key := int64(math.MinInt64) // int64(-∞)   (qualifies for ≤)
+		key := int64(_LKeyMin) // int64(-∞)   (qualifies for ≤)
 		if i > 0 {
 			// key[0] is unused and not saved
 			key, ok = t[idx].(int64) // XXX Xint
@@ -611,7 +662,7 @@ func (bt *lobtreeState) PySetState(pystate interface{}) (err error) {
 		}
 
 		if i > 1 && !(key > kprev) {
-			fmt.Errorf("data: [%d]: key not ↑", i)
+			return fmt.Errorf("data: [%d]: key not ↑", i)
 		}
 		kprev = key
 
@@ -631,7 +682,7 @@ func (bt *lobtreeState) PySetState(pystate interface{}) (err error) {
 			childrenKind = kind
 		}
 		if kind != childrenKind {
-			fmt.Errorf("data: [%d]: children must be of the same type", i)
+			return fmt.Errorf("data: [%d]: children must be of the same type", i)
 		}
 
 		bt.data = append(bt.data, LOEntry{key: kkey, child: child.(LONode)})
@@ -648,3 +699,60 @@ func init() {
 	zodb.RegisterClass("BTrees.LOBTree.LOBTree",  t(LOBTree{}),  t(lobtreeState{}))
 	zodb.RegisterClass("BTrees.LOBTree.LOBucket", t(LOBucket{}), t(lobucketState{}))
 }
+
+
+// ---- misc ----
+
+// Has returns whether key k belongs to the range.
+func (r *LKeyRange) Has(k int64) bool {
+	return (r.Lo <= k && k <= r.Hi_)
+}
+
+// Empty returns whether key range is empty.
+func (r *LKeyRange) Empty() bool {
+	hi := r.Hi_
+	if hi == _LKeyMax {
+		// [x,∞] cannot be empty because max x is ∞ and [∞,∞] has one element: ∞
+		return false
+	}
+	hi++ // no overflow
+	return r.Lo >= hi
+}
+
+func (r LKeyRange) String() string {
+	var shi string
+	if r.Hi_ == _LKeyMax {
+		shi = lkstr(r.Hi_) // ∞
+	} else {
+		shi = fmt.Sprintf("%d", r.Hi_+1)
+	}
+	return fmt.Sprintf("[%s,%s)", lkstr(r.Lo), shi)
+}
+
+
+func lkmin(a, b int64) int64 {
+	if a < b {
+		return a
+	} else {
+		return b
+	}
+}
+
+func lkmax(a, b int64) int64 {
+	if a > b {
+		return a
+	} else {
+		return b
+	}
+}
+
+// lkstr formats key as string.
+func lkstr(k int64) string {
+	if k == _LKeyMin {
+		return "-∞"
+	}
+	if k == _LKeyMax {
+		return "∞"
+	}
+	return fmt.Sprintf("%d", k)
+}

go/zodb/btree/ztestdata_expect_test.go

@@ -3,13 +3,24 @@ package btree
 
 
 var smallTestv = [...]testEntry{
-	testEntry{oid: 6, kind: kindBucket, itemv: []kv{}},
-	testEntry{oid: 3, kind: kindBucket, itemv: []kv{{10, int64(17)}, }},
+	testEntry{oid: 7, kind: kindBucket, itemv: []kv{}},
+	testEntry{oid: 4, kind: kindBucket, itemv: []kv{{10, int64(17)}, }},
 	testEntry{oid: 1, kind: kindBucket, itemv: []kv{{15, int64(1)}, {23, "hello"}, }},
-	testEntry{oid: 2, kind: kindBTree, itemv: []kv{}},
-	testEntry{oid: 7, kind: kindBTree, itemv: []kv{{5, int64(4)}, }},
-	testEntry{oid: 4, kind: kindBTree, itemv: []kv{{7, int64(3)}, {9, "world"}, }},
+	testEntry{oid: 3, kind: kindBTree, itemv: []kv{}},
+	testEntry{oid: 8, kind: kindBTree, itemv: []kv{{5, int64(4)}, }},
+	testEntry{oid: 5, kind: kindBTree, itemv: []kv{{7, int64(3)}, {9, "world"}, }},
 }
 
-const B3_oid = 5
+const B3_oid = 6
 const B3_maxkey = 9999
+
+const Bv_oid = 2
+const Bv_kmin = 1
+const Bv_kmax = 9
+var Bvdict = map[int64][]tVisit{
+	1: []tVisit{{2, LKeyRange{_LKeyMin, _LKeyMax}}, {342, LKeyRange{_LKeyMin, 3}}, {344, LKeyRange{_LKeyMin, 1}}},
+	2: []tVisit{{2, LKeyRange{_LKeyMin, _LKeyMax}}, {342, LKeyRange{_LKeyMin, 3}}, {349, LKeyRange{2, 3}}},
+	5: []tVisit{{2, LKeyRange{_LKeyMin, _LKeyMax}}, {343, LKeyRange{4, _LKeyMax}}, {345, LKeyRange{4, _LKeyMax}}, {346, LKeyRange{4, 6}}},
+	8: []tVisit{{2, LKeyRange{_LKeyMin, _LKeyMax}}, {343, LKeyRange{4, _LKeyMax}}, {345, LKeyRange{4, _LKeyMax}}, {347, LKeyRange{7, 8}}},
+	9: []tVisit{{2, LKeyRange{_LKeyMin, _LKeyMax}}, {343, LKeyRange{4, _LKeyMax}}, {345, LKeyRange{4, _LKeyMax}}, {348, LKeyRange{9, _LKeyMax}}},
+}

go/zodb/connection.go

-// Copyright (C) 2018-2020  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -180,7 +180,7 @@ const (
 
 // newConnection creates new Connection associated with db.
 func newConnection(db *DB, at Tid) *Connection {
-	return &Connection{
+	conn := &Connection{
 		db:    db,
 		at:    at,
 		cache: LiveCache{
@@ -188,6 +188,10 @@ func newConnection(db *DB, at Tid) *Connection {
 			objtab: make(map[Oid]*weak.Ref),
 		},
 	}
+	if cc := db.opt.CacheControl; cc != nil {
+		conn.cache.SetControl(cc)
+	}
+	return conn
 }
 
 // DB returns database handle under which the connection was opened.

go/zodb/db.go

-// Copyright (C) 2018-2020  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -48,6 +48,8 @@ type DB struct {
 	stor   IStorage
 	watchq chan Event // we are watching .stor via here
 
+	opt *DBOptions
+
 	down     chan struct{} // ready when DB is no longer operational
 	downOnce sync.Once     // shutdown may be due to both Close and IO error in watcher
 	downErr  error         // reason for shutdown
@@ -121,15 +123,23 @@ type DB struct {
 	// (so it is not duplicated many times for many DB case)
 }
 
+// DBOptions describes options to NewDB.
+type DBOptions struct {
+	// CacheControl, if !nil, is set as default live cache control for
+	// newly created connections.
+	CacheControl LiveCacheControl
+}
 
 // NewDB creates new database handle.
 //
 // Created database handle must be closed when no longer needed.
-func NewDB(stor IStorage) *DB {
-	// XXX db options?
+func NewDB(stor IStorage, opt *DBOptions) *DB {
+	// copy opts in case caller will change them later
+	opt_ := *opt
 	db := &DB{
 		stor:   stor,
 		watchq: make(chan Event),
+		opt:    &opt_,
 		down:   make(chan struct{}),
 		hwait:  make(map[hwaiter]struct{}),
 

go/zodb/impexp_test.go

-// Copyright (C) 2019  Nexedi SA and Contributors.
-//                     Kirill Smelkov <kirr@nexedi.com>
+// Copyright (C) 2019-2021  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
@@ -25,10 +25,16 @@ import (
 
 // imported at runtime via import_x_test due to cyclic dependency:
 
-var ZPyCommit func(string, Tid, ...IPersistent) (Tid, error)
+var ZPyCommit    func(string, Tid, ...IPersistent) (Tid, error)
+var ZPyCommitRaw func(string, Tid, ...ZRawObject)  (Tid, error)
 
 // exported for zodb_test package:
 
+type ZRawObject struct { // keep in sync with xtesting.ZRawObject
+	Oid  Oid
+	Data []byte
+}
+
 func PSerialize(obj IPersistent) *mem.Buf {
 	return obj.persistent().pSerialize()
 }

go/zodb/import_x_test.go

-// Copyright (C) 2019  Nexedi SA and Contributors.
-//                     Kirill Smelkov <kirr@nexedi.com>
+// Copyright (C) 2019-2021  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
@@ -33,7 +33,8 @@ import (
 // due to cyclic dependency.
 
 func init() {
-	zodb.ZPyCommit = ZPyCommit
+	zodb.ZPyCommit    = ZPyCommit
+	zodb.ZPyCommitRaw = ZPyCommitRaw
 }
 
 // ZPyCommit commits new transaction with specified objects.
@@ -41,8 +42,8 @@ func init() {
 // The objects need to be alive, but do not need to be marked as changed.
 // The commit is performed via zodb/py.
 func ZPyCommit(zurl string, at zodb.Tid, objv ...zodb.IPersistent) (zodb.Tid, error) {
-	var rawobjv []xtesting.ZRawObject // raw zodb objects data to commit
-	var bufv []*mem.Buf               // buffers to release
+	var rawobjv []zodb.ZRawObject // raw zodb objects data to commit
+	var bufv []*mem.Buf           // buffers to release
 	defer func() {
 		for _, buf := range bufv {
 			buf.Release()
@@ -51,7 +52,7 @@ func ZPyCommit(zurl string, at zodb.Tid, objv ...zodb.IPersistent) (zodb.Tid, er
 
 	for _, obj := range objv {
 		buf := zodb.PSerialize(obj)
-		rawobj := xtesting.ZRawObject{
+		rawobj := zodb.ZRawObject{
 			Oid:  obj.POid(),
 			Data: buf.Data,
 		}
@@ -59,5 +60,13 @@ func ZPyCommit(zurl string, at zodb.Tid, objv ...zodb.IPersistent) (zodb.Tid, er
 		bufv = append(bufv, buf)
 	}
 
-	return xtesting.ZPyCommitRaw(zurl, at, rawobjv...)
+	return ZPyCommitRaw(zurl, at, rawobjv...)
+}
+
+func ZPyCommitRaw(zurl string, at zodb.Tid, rawobjv ...zodb.ZRawObject) (zodb.Tid, error) {
+	var xrawobjv []xtesting.ZRawObject
+	for _, obj := range rawobjv {
+		xrawobjv = append(xrawobjv, xtesting.ZRawObject{Oid: obj.Oid, Data: obj.Data})
+	}
+	return xtesting.ZPyCommitRaw(zurl, at, xrawobjv...)
 }

go/zodb/persistent.go

-// Copyright (C) 2018-2019  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -163,7 +163,7 @@ func (obj *Persistent) pSerialize() *mem.Buf {
 func (obj *Persistent) PActivate(ctx context.Context) (err error) {
 	obj.mu.Lock()
 	obj.refcnt++
-	doload := (obj.refcnt == 1 && obj.state == GHOST)
+	doload := (obj.state == GHOST && obj.loading == nil)
 	defer func() {
 		if err != nil {
 			obj.PDeactivate()
@@ -208,10 +208,10 @@ func (obj *Persistent) PActivate(ctx context.Context) (err error) {
 			"%v (want %v); .loading = %p (want %p)", s, GHOST, l, loading))
 	}
 
-	obj.serial = serial
-
 	// try to pass loaded state to object
 	if err == nil {
+		obj.serial = serial
+
 		switch istate := obj.istate().(type) {
 		case Stateful:
 			err = istate.SetState(state)
@@ -229,9 +229,13 @@ func (obj *Persistent) PActivate(ctx context.Context) (err error) {
 		if err == nil {
 			obj.state = UPTODATE
 		}
+
+	} else {
+		obj.serial = InvalidTid
+		// force reload on next activate if it was an error
+		obj.loading = nil
 	}
 
-	// XXX set state to load error? (to avoid panic on second activate after load error)
 	loading.err = err
 
 	obj.mu.Unlock()
@@ -257,7 +261,7 @@ func (obj *Persistent) PDeactivate() {
 	if obj.state >= CHANGED {
 		return
 	}
-	if obj.oid == InvalidOid { // newly created not-yet committed object	// TODO tests
+	if obj.oid == InvalidOid { // newly created not-yet committed object
 		return
 	}
 

go/zodb/persistent_test.go

-// Copyright (C) 2018-2020  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -21,6 +21,7 @@ package zodb
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -276,7 +277,7 @@ func (t *tDB) Reopen() {
 		ReadOnly: true,
 		NoCache: !t.rawcache,
 	}); X(err)
-	db := NewDB(stor)
+	db := NewDB(stor, &DBOptions{})
 	t.stor = stor
 	t.db = db
 }
@@ -317,7 +318,7 @@ func (t *tDB) Add(oid Oid, value string) {
 }
 
 // Commit commits objects queued by Add.
-func (t *tDB) Commit() {
+func (t *tDB) Commit() Tid {
 	t.Helper()
 
 	head, err := ZPyCommit(t.zurl, t.head, t.commitq...)
@@ -326,6 +327,19 @@ func (t *tDB) Commit() {
 	}
 	t.head = head
 	t.commitq = nil
+	return head
+}
+
+// CommitRaw commits raw changes.
+func (t *tDB) CommitRaw(rawobjv ...ZRawObject) Tid {
+	t.Helper()
+
+	head, err := ZPyCommitRaw(t.zurl, t.head, rawobjv...)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.head = head
+	return head
 }
 
 // Open opens new test transaction/connection.
@@ -443,12 +457,14 @@ func (t *tConnection) Abort() {
 }
 
 func (t *tDB) fatalif(err error) {
+	t.Helper()
 	if err != nil {
 		t.Fatal(err)
 	}
 }
 
 func (t *tConnection) fatalif(err error) {
+	t.Helper()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -473,13 +489,11 @@ func testPersistentDB(t0 *testing.T, rawcache bool) {
 
 	tdb.Add(101, "bonjour")
 	tdb.Add(102, "monde")
-	tdb.Commit()
-	at0 := tdb.head
+	at0 := tdb.Commit()
 
 	tdb.Add(101, "hello")
 	tdb.Add(102, "world")
-	tdb.Commit()
-	at1 := tdb.head
+	at1 := tdb.Commit()
 
 	tdb.Reopen() // so that at0 is not covered by db.δtail
 	db := tdb.db
@@ -541,8 +555,7 @@ func testPersistentDB(t0 *testing.T, rawcache bool) {
 
 	// commit change to obj2 from external process
 	tdb.Add(102, "kitty")
-	tdb.Commit()
-	at2 := tdb.head
+	at2 := tdb.Commit()
 
 	// new db connection should see the change
 	t2 := tdb.Open(&ConnOptions{})
@@ -608,6 +621,18 @@ func testPersistentDB(t0 *testing.T, rawcache bool) {
 	t.checkObj(obj2, 102, at2, UPTODATE,    0, "kitty")
 
 
+	// newly created object should not go to ghost on deactivate
+	obj3 := NewMyObject(t.conn)
+	obj3.value = "new"
+	checkObj(t0, obj3, t.conn, InvalidOid, InvalidTid, UPTODATE, 0)
+	assert.Equal(obj3.value, "new")
+	t.PActivate(obj3)
+	checkObj(t0, obj3, t.conn, InvalidOid, InvalidTid, UPTODATE, 1)
+	obj3.PDeactivate()
+	checkObj(t0, obj3, t.conn, InvalidOid, InvalidTid, UPTODATE, 0)
+	assert.Equal(obj3.value, "new")
+
+
 	// finish tnx3 and txn2 - conn1 and conn2 go back to db pool
 	t.Abort()
 	t2.Abort()
@@ -719,6 +744,69 @@ func testPersistentDB(t0 *testing.T, rawcache bool) {
 	t.checkObj(robj2, 102, InvalidTid, GHOST, 0)
 }
 
+// Verify that PActivate works correctly after hitting an error from the storage.
+// In this test the error is "object was deleted".
+func TestActivateAfterDelete(t0 *testing.T) {
+	assert := assert.New(t0)
+
+	tdb := testdb(t0, /*rawcache=*/false)
+	defer tdb.Close()
+	db := tdb.db
+
+	tdb.Add(101, "object")
+	at0 := tdb.Commit()
+
+	t := tdb.Open(&ConnOptions{})
+
+	// do not evict the object from live cache.
+	zcc := &zcacheControl{map[Oid]PCachePolicy{
+		101: PCachePinObject | PCacheKeepState,
+	}}
+	zcache := t.conn.Cache()
+	zcache.Lock()
+	zcache.SetControl(zcc)
+	zcache.Unlock()
+
+	// load the object
+	obj := t.Get(101)
+	t.checkObj(obj, 101, InvalidTid, GHOST, 0)
+	t.PActivate(obj)
+	t.checkObj(obj, 101, at0, UPTODATE, 1, "object")
+	obj.PDeactivate()
+	t.checkObj(obj, 101, at0, UPTODATE, 0, "object")
+
+	// delete obj
+	at1 := tdb.CommitRaw(ZRawObject{Oid: 101, Data: nil})
+
+	// conn stays at older view with obj pinned into the cache
+	t.checkObj(obj, 101, at0, UPTODATE, 0, "object")
+
+	// finish transaction and reopen new connection - it should be the same conn
+	t.Abort()
+	assert.Equal(db.pool, []*Connection{t.conn})
+	t_ := tdb.Open(&ConnOptions{})
+	assert.Same(t_.conn, t.conn)
+	t = t_
+	assert.Equal(t.conn.At(), at1)
+
+	// obj should be invalidated but present in the cache
+	t.checkObj(obj, 101, InvalidTid, GHOST, 0)
+
+	// activating obj should give "no data" error
+	// loop because second activate used to panic
+	for i := 0; i < 10; i++ {
+		err := obj.PActivate(t.ctx)
+		eok := &NoDataError{Oid: obj.POid(), DeletedAt: at1}
+		var e *NoDataError
+		errors.As(err, &e)
+		if !reflect.DeepEqual(e, eok) {
+			t.Fatalf("(%d) after delete: err:\nhave: %s\nwant cause: %s", i, err, eok)
+		}
+		// obj should stay in the cache in ghost state
+		t.checkObj(obj, 101, InvalidTid, GHOST, 0)
+	}
+}
+
 // Test details of how LiveCache handles live caching policy.
 func TestLiveCache(t0 *testing.T) {
 	assert := assert.New(t0)
@@ -730,8 +818,7 @@ func TestLiveCache(t0 *testing.T) {
 	tdb.Add(102, "труд")
 	tdb.Add(103, "май")
 	tdb.Add(104, "весна")
-	tdb.Commit()
-	at1 := tdb.head
+	at1 := tdb.Commit()
 
 	zcc := &zcacheControl{map[Oid]PCachePolicy{
 		// obj1 - default (currently: don't pin and don't keep state)

go/zodb/py/pydata-gen-testdata

 #!/usr/bin/env python2
 # -*- coding: utf-8 -*-
-# Copyright (C) 2017-2020  Nexedi SA and Contributors.
+# Copyright (C) 2017-2021  Nexedi SA and Contributors.
 #                          Kirill Smelkov <kirr@nexedi.com>
 #
 # This program is free software: you can Use, Study, Modify and Redistribute
@@ -21,7 +21,7 @@
 """generate reference pickle objects encoding for tests"""
 
 from ZODB import serialize
-from zodbtools.test.gen_testdata import run_with_zodb3py2_compat
+from zodbtools.test.gen_testdata import run_with_zodb4py2_compat
 from golang.gcompat import qq
 
 def main2():
@@ -56,7 +56,7 @@ def main2():
         emit("}")
 
 def main():
-    run_with_zodb3py2_compat(main2)
+    run_with_zodb4py2_compat(main2)
 
 if __name__ == '__main__':
     main()

go/zodb/storage.go

-// Copyright (C) 2017-2019  Nexedi SA and Contributors.
+// Copyright (C) 2017-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -46,12 +46,30 @@ type DriverOptions struct {
 	//
 	// Watchq can be nil to ignore such events. However if Watchq != nil, the events
 	// have to be consumed or else the storage driver will misbehave - e.g.
-	// it can get out of sync with the on-disk database file.
+	// it can get out of sync with the on-disk database file, or deadlock
+	// on any user-called operation.
 	//
 	// The storage driver closes !nil Watchq when the driver is closed.
 	//
 	// The storage driver will send only and all events in (at₀, +∞] range,
 	// where at₀ is at returned by driver open.
+	//
+	// The storage driver will stop sending events after call to Close.
+	// In particular the following example is valid and safe from deadlock:
+	//
+	//	watchq := make(chan zodb.Event)
+	//	stor, at0, err := zodb.OpenDriver(..., &DriverOptions{Watchq: watchq})
+	//	defer stor.Close()
+	//
+	//	for {
+	//		select {
+	//		case <-ctx.Done():
+	//			return ctx.Err()
+	//
+	//		case <-watchq:
+	//			...
+	//		}
+	//	}
 	Watchq chan<- Event
 }
 
@@ -73,10 +91,13 @@ func RegisterDriver(scheme string, opener DriverOpener) {
 	driverRegistry[scheme] = opener
 }
 
-// XXX
-func openDriver(ctx context.Context, zurl string, opt *DriverOptions) (_ IStorageDriver, at0 Tid, _ error) {
+// OpenDriver opens ZODB storage driver by URL.
+//
+// It is similar to Open but returns low-level IStorageDriver instead of IStorage.
+// Most users should use Open.
+func OpenDriver(ctx context.Context, zurl string, opt *DriverOptions) (_ IStorageDriver, at0 Tid, _ error) {
 	// no scheme -> file://
-	if !strings.Contains(zurl, "://") {
+	if !strings.Contains(zurl, ":") {
 		zurl = "file://" + zurl
 	}
 
@@ -91,7 +112,7 @@ func openDriver(ctx context.Context, zurl string, opt *DriverOptions) (_ IStorag
 
 	opener, ok := driverRegistry[u.Scheme]
 	if !ok {
-		return nil, InvalidTid, fmt.Errorf("zodb: URL scheme \"%s://\" not supported", u.Scheme)
+		return nil, InvalidTid, fmt.Errorf("zodb: URL scheme \"%s:\" not supported", u.Scheme)
 	}
 
 	storDriver, at0, err := opener(ctx, u, opt)
@@ -108,7 +129,7 @@ func openDriver(ctx context.Context, zurl string, opt *DriverOptions) (_ IStorag
 // Users should import in storage packages they use or zodb/wks package to
 // get support for well-known storages.
 //
-// Storage authors should register their storages with RegisterStorage.
+// Storage authors should register their storages with RegisterDriver.
 func Open(ctx context.Context, zurl string, opt *OpenOptions) (IStorage, error) {
 	drvWatchq := make(chan Event)
 	drvOpt := &DriverOptions{
@@ -116,7 +137,7 @@ func Open(ctx context.Context, zurl string, opt *OpenOptions) (IStorage, error)
 		Watchq:   drvWatchq,
 	}
 
-	storDriver, at0, err := openDriver(ctx, zurl, drvOpt)
+	storDriver, at0, err := OpenDriver(ctx, zurl, drvOpt)
 	if err != nil {
 		return nil, err
 	}
@@ -160,9 +181,10 @@ type storage struct {
 	driver  IStorageDriver
 	l1cache *Cache // can be =nil, if opened with NoCache
 
-	down     chan struct{} // ready when no longer operational
-	downOnce sync.Once     // shutdown may be due to both Close and IO error in watcher|Sync
-	downErr  error         // reason for shutdown
+	down        chan struct{} // ready when no longer operational
+	downOnce    sync.Once     // shutdown may be due to both Close and IO error in watcher|Sync
+	downErr     error         // reason for shutdown
+	drvCloseErr error         // err from .driver.Close()
 
 	// watcher
 
@@ -185,6 +207,11 @@ func (s *storage) shutdown(reason error) {
 	s.downOnce.Do(func() {
 		close(s.down)
 		s.downErr = fmt.Errorf("not operational due: %s", reason)
+		// - if called by Close or failed Sync: driver.Close will close
+		//   drvWatchq and cause watcher to stop.
+		// - if called by failed watcher: closing driver will prevent
+		//   drvWatchq<- deadlock in driver because we no longer read from it.
+		s.drvCloseErr = s.driver.Close()
 	})
 }
 
@@ -198,7 +225,7 @@ func (s *storage) Iterate(ctx context.Context, tidMin, tidMax Tid) ITxnIterator
 
 func (s *storage) Close() error {
 	s.shutdown(fmt.Errorf("closed"))
-	return s.driver.Close() // this will close drvWatchq and cause watcher stop
+	return s.drvCloseErr
 }
 
 // loading goes through cache - this way prefetching can work
@@ -369,7 +396,7 @@ func (s *storage) _watcher() error {
 func (s *storage) AddWatch(watchq chan<- Event) (at0 Tid) {
 	ack := make(chan Tid)
 	select {
-	// no longer operational: behave if watchq was registered before that
+	// no longer operational: behave as if watchq was registered before that
 	// and then seen down/close events. Interact with DelWatch directly.
 	case <-s.down:
 		s.headMu.Lock()   // shutdown may be due to Close call and watcher might be
@@ -471,6 +498,14 @@ func (s *storage) Sync(ctx context.Context) (err error) {
 	}
 
 	// wait till .head >= head
+	// XXX instead require from drivers that `drv.Sync() -> head`
+	// guarantees that all EventCommit with .tid <= head were sent to
+	// watchq
+	//
+	// https://lab.nexedi.com/nexedi/ZODB/commit/40116375
+	// https://github.com/zopefoundation/ZODB/commit/4a6b0283#diff-d2a01f71a79ac2b379e218cf72fa1205d3426cad19e7b72d71899f643be4bb73
+	//
+	// ?
 	watchq := make(chan Event)
 	at = s.AddWatch(watchq)
 	defer s.DelWatch(watchq)

go/zodb/storage/demo/demo_test.go

+// Copyright (C) 2021  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.
+
+package demo
+
+import (
+	"bytes"
+	"context"
+	"io/ioutil"
+	"fmt"
+	"net/url"
+	"os"
+	"reflect"
+	"regexp"
+	"testing"
+
+	"lab.nexedi.com/kirr/go123/xerr"
+
+	"lab.nexedi.com/kirr/neo/go/internal/xtesting"
+	"lab.nexedi.com/kirr/neo/go/zodb"
+	"lab.nexedi.com/kirr/neo/go/zodb/zodbtools"
+
+	// for file: scheme support
+	_ "lab.nexedi.com/kirr/neo/go/zodb/storage/fs1"
+)
+
+// DemoData represents data for a demo: storage.
+type DemoData struct {
+	base string // url for   base.fs
+	δ    string // ----/---- δ.fs
+}
+
+func (ddat *DemoData) URL() string {
+	return fmt.Sprintf("demo:(%s)/(%s)", ddat.base, ddat.δ)
+}
+
+
+// tOptions represents options for testing.
+// TODO -> xtesting
+type tOptions struct {
+	Preload string // preload database with data from this location
+}
+
+// withDemoData tests f with all kinds of opt.Preload data split into base + δ.
+func withDemoData(t *testing.T, f func(t *testing.T, ddat *DemoData), optv ...tOptions) {
+	t.Helper()
+	X := xtesting.FatalIf(t)
+
+	opt := tOptions{}
+	if len(optv) > 1 {
+		panic("multiple tOptions not allowed")
+	}
+	if len(optv) == 1 {
+		opt = optv[0]
+	}
+
+	// retrieve zdump of Preload
+	zdump := ""
+	if opt.Preload != "" {
+		ctx := context.Background()
+		buf := &bytes.Buffer{}
+		stor, err := zodb.Open(ctx, opt.Preload, &zodb.OpenOptions{ReadOnly: true}); X(err)
+		err = zodbtools.Dump(ctx, buf, stor, 0, zodb.TidMax, /*hashonly=*/false)
+		stor.Close()
+		X(err)
+		zdump = buf.String()
+	}
+
+	// split zdump into transactions
+	// XXX hacky; TODO -> zodbtools.DumpReader
+	txnRe := regexp.MustCompile(`(?m)^txn (?P<tid>[0-9a-f]{16}) "(?P<status>.)"$`)
+	type zdumpTxn struct {
+		tid zodb.Tid
+		pos int      // where this transaction starts in the dump
+	}
+	var txnv []zdumpTxn
+	for _, m := range txnRe.FindAllStringSubmatchIndex(zdump, -1) {
+		// [m[0]:m[1]] refers to whole txn line
+		__ := zdump[m[2]:m[3]]
+		tid, err := zodb.ParseTid(__); X(err)
+		txnv = append(txnv, zdumpTxn{tid, m[0]})
+	}
+
+	// verify f on all combinations of preload being split into base+δ
+	work := xtempdir(t)
+	defer os.RemoveAll(work)
+	test1 := func(δstart zodb.Tid, zdumpBase, zdumpδ string) {
+		t.Helper()
+		t.Run(fmt.Sprintf("δstart=%s", δstart), func(t *testing.T) {
+			t.Helper()
+			X := xtesting.FatalIf(t)
+			work1 := work + "/δ" + δstart.String()
+			err := os.Mkdir(work1, 0777); X(err)
+
+			base := "file://"+work1+"/base.fs"
+			δ    := "file://"+work1+"/δ.fs"
+			ddat := &DemoData{base, δ}
+
+			_, err = xtesting.ZPyRestore(base, zdumpBase);  X(err)
+			// restore δ part via `demo:(base)/(δ)` - not `file:δ`.
+			// The reason we do this is because restoring δ via
+			// just its file will fail when restoring copy data
+			// record with copy_from transaction being in base.
+			_, err = xtesting.ZPyRestore(ddat.URL(), zdumpδ);  X(err)
+
+			f(t, ddat)
+		})
+	}
+	for i := 0; i < len(txnv); i++ {
+		δtail := txnv[i]
+		test1(δtail.tid, zdump[:δtail.pos], zdump[δtail.pos:])
+	}
+	test1(zodb.TidMax, zdump, "")
+}
+
+// withDemo tests f with demo: client connected to all kind of demo data splits.
+func withDemo(t *testing.T, f func(t *testing.T, ddat *DemoData, ddrv *Storage), optv ...tOptions) {
+	t.Helper()
+	withDemoData(t, func(t *testing.T, ddat *DemoData) {
+		t.Helper()
+		X := xtesting.FatalIf(t)
+		ddrv, _, err := demoOpen(ddat.URL(), &zodb.DriverOptions{ReadOnly: true}); X(err)
+		defer func() {
+			err := ddrv.Close(); X(err)
+		}()
+
+		f(t, ddat, ddrv)
+	}, optv...)
+}
+
+func TestURL(t *testing.T) {
+	withDemo(t, func(t *testing.T, ddat *DemoData, ddrv *Storage) {
+		zurl   := ddrv.URL()
+		zurlOk := ddat.URL()
+		if zurl != zurlOk {
+			t.Fatalf("bad zurl:\nhave: %s\nwant: %s", zurl, zurlOk)
+		}
+	})
+}
+
+func TestEmptyDB(t *testing.T) {
+	withDemo(t, func(t *testing.T, _ *DemoData, ddrv *Storage) {
+		xtesting.DrvTestEmptyDB(t, ddrv)
+	})
+}
+
+func TestLoad(t *testing.T) {
+	X := xtesting.FatalIf(t)
+
+	data := "../fs1/testdata/1.fs"
+	txnvOk, err := xtesting.LoadDBHistory(data); X(err)
+
+	withDemo(t, func(t *testing.T, _ *DemoData, ddrv *Storage) {
+		xtesting.DrvTestLoad(t, ddrv, txnvOk)
+	}, tOptions{
+		Preload: data,
+	})
+}
+
+func TestWatch(t *testing.T) {
+	withDemoData(t, func(t *testing.T, ddat *DemoData) {
+		xtesting.DrvTestWatch(t, ddat.URL(), openByURL)
+	})
+}
+
+// MutateBase mutates ddat.base with new commit.
+func (ddat *DemoData) MutateBase() (zodb.Tid, error) {
+	return xtesting.ZPyCommitRaw(ddat.base, 0, xtesting.ZRawObject{
+		Oid:  1,
+		Data: []byte("ZZZ"),
+	})
+}
+
+// TestSync_vs_BaseMutate verifies Sync wrt base mutation.
+func TestSync_vs_BaseMutate(t *testing.T) {
+	withDemo(t, func(t *testing.T, ddat *DemoData, ddrv *Storage) {
+		X := xtesting.FatalIf(t)
+
+		head, err := ddrv.Sync(context.Background())
+		if !(head == 0 && err == nil) {
+			t.Fatalf("sync0: head=%s  err=%s", head, err)
+		}
+
+		tid, err := ddat.MutateBase(); X(err)
+
+		head, err = ddrv.Sync(context.Background())
+		errOk := &zodb.OpError{URL: ddrv.URL(), Op: "sync", Err: &baseMutatedError{
+			baseAt0:  0,
+			baseHead: tid,
+		}}
+		if !reflect.DeepEqual(err, errOk) {
+			t.Fatalf("after base mutate: sync: unexpected error:\nhave: %s\nwant: %s",
+				err, errOk)
+		}
+	})
+}
+
+// TestWatchLoad_vs_BaseMutate verifies Watch and Load wrt base mutation.
+func TestWatchLoad_vs_BaseMutate(t *testing.T) {
+	withDemoData(t, func(t *testing.T, ddat *DemoData) {
+		X := xtesting.FatalIf(t)
+
+		watchq := make(chan zodb.Event)
+		ddrv, at0, err := demoOpen(ddat.URL(), &zodb.DriverOptions{
+			ReadOnly: true,
+			Watchq:   watchq,
+		}); X(err)
+		defer func() {
+			err := ddrv.Close(); X(err)
+		}()
+
+		tid, err := ddat.MutateBase(); X(err)
+
+		// first wait for error from watchq
+		event, ok := <-watchq
+		if !ok {
+			t.Fatal("after base mutate: premature watchq close")
+		}
+		evErr, ok := event.(*zodb.EventError)
+		if !ok {
+			t.Fatalf("after base mutate: unexpected event: %T", event)
+		}
+
+		errBaseMutated := &baseMutatedError{
+			baseAt0:  0,
+			baseHead: tid,
+		}
+		evErrOk := &zodb.EventError{&zodb.OpError{URL: ddrv.URL(), Op: "watcher", Err: errBaseMutated}}
+		if !reflect.DeepEqual(evErr, evErrOk) {
+			t.Fatalf("after base mutate: unexpected event:\nhave: %s\nwant: %s", evErr, evErrOk)
+		}
+
+		// now make sure Load fails with "base mutated" error
+		xid := zodb.Xid{Oid: 1, At: at0}
+		data, serial, err := ddrv.Load(context.Background(), xid)
+		errOk := &zodb.OpError{URL: ddrv.URL(), Op: "load", Args: xid, Err: errBaseMutated}
+		if !reflect.DeepEqual(err, errOk) {
+			t.Fatalf("after base mutate: load: unexpected error:\nhave: %s\nwant: %s",
+				err, errOk)
+		}
+		if !(data == nil && serial == zodb.InvalidTid) {
+			t.Fatalf("after base mutate: load: unexpected data=%v  serial=%v", data, serial)
+		}
+	})
+}
+
+
+func demoOpen(zurl string, opt *zodb.DriverOptions) (_ *Storage, at0 zodb.Tid, err error) {
+	defer xerr.Contextf(&err, "opendemo %s", zurl)
+	u, err := url.Parse(zurl)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	d, at0, err := openByURL(context.Background(), u, opt)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	return d.(*Storage), at0, nil
+}
+
+func xtempdir(t *testing.T) string {
+	t.Helper()
+	tmpd, err := ioutil.TempDir("", "demo")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return tmpd
+}

go/zodb/storage/fs1/filestorage.go

-// Copyright (C) 2017-2020  Nexedi SA and Contributors.
+// Copyright (C) 2017-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -109,6 +109,9 @@ type FileStorage struct {
 	downOnce sync.Once      // shutdown may be due to both Close and IO error in watcher
 	errClose error          // error from .file.Close()
 	watchWg  sync.WaitGroup // to wait for watcher finish
+
+	closed    chan struct{} // ready when storage was Closed
+	closeOnce sync.Once
 }
 
 // IStorageDriver
@@ -132,7 +135,7 @@ func (fs *FileStorage) LastOid(_ context.Context) (zodb.Oid, error) {
 }
 
 func (fs *FileStorage) URL() string {
-	return fs.file.Name()
+	return "file://" + fs.file.Name()
 }
 
 // freelist(DataHeader)
@@ -468,7 +471,7 @@ func (fs *FileStorage) watcher(w *fsnotify.Watcher, errFirstRead chan<- error) {
 		// XXX it can also be internal.poll.ErrFileClosing
 		e.Err.Error() == "use of closed file") {
 		select {
-		case <-fs.down:
+		case <-fs.closed:
 			err = nil
 		default:
 		}
@@ -485,7 +488,13 @@ func (fs *FileStorage) watcher(w *fsnotify.Watcher, errFirstRead chan<- error) {
 
 	if fs.watchq != nil {
 		if err != nil {
-			fs.watchq <- &zodb.EventError{err}
+			select {
+			case <-fs.closed:
+				// closed - skip send to watchq
+
+			case fs.watchq <- &zodb.EventError{err}:
+				// ok
+			}
 		}
 		close(fs.watchq)
 	}
@@ -535,9 +544,9 @@ mainloop:
 		if !first {
 			traceWatch("select ...")
 			select {
-			case <-fs.down:
+			case <-fs.closed:
 				// closed
-				traceWatch("down")
+				traceWatch("closed")
 				return nil
 
 			case err := <-w.Errors:
@@ -694,7 +703,7 @@ mainloop:
 			// notify client
 			if fs.watchq != nil {
 				select {
-				case <-fs.down:
+				case <-fs.closed:
 					return nil
 
 				case fs.watchq <- &zodb.EventCommit{it.Txnh.Tid, δoid}:
@@ -772,6 +781,9 @@ func (fs *FileStorage) shutdown(reason error) {
 }
 
 func (fs *FileStorage) Close() error {
+	fs.closeOnce.Do(func() {
+		close(fs.closed)
+	})
 	fs.shutdown(fmt.Errorf("closed"))
 	fs.watchWg.Wait()
 
@@ -794,6 +806,7 @@ func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileSto
 	fs := &FileStorage{
 		watchq: opt.Watchq,
 		down:   make(chan struct{}),
+		closed: make(chan struct{}),
 	}
 
 	f, err := os.Open(path)

go/zodb/storage/fs1/filestorage_test.go

-// Copyright (C) 2017-2020  Nexedi SA and Contributors.
+// Copyright (C) 2017-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -370,3 +370,37 @@ func TestOpenRecovery(t *testing.T) {
 		})
 	}
 }
+
+
+// TestLoadWhiteout verifies access to whiteout data record.
+//
+// Whiteout is data record with deletion when object was not previously there.
+// It has both len(data)=0 and backpointer=0.
+//
+// TODO merge into regular tests on testdata/1.fs when
+// FileStorage/py.deleteObject allows to create whiteouts instead of raising
+// POSKeyError.
+func TestLoadWhiteout(t *testing.T) {
+	fs, _ := xfsopen(t, "testdata/whiteout.fs")
+	defer exc.XRun(fs.Close)
+
+	xid := zodb.Xid{At: zodb.Tid(0x17), Oid: zodb.Oid(1)}
+	buf, serial, err := fs.Load(context.Background(), xid)
+
+	errOk := &zodb.OpError{
+		URL:  fs.URL(),
+		Op:   "load",
+		Args: xid,
+		Err:  &zodb.NoDataError{Oid: xid.Oid, DeletedAt: xid.At},
+	}
+
+	if !reflect.DeepEqual(err, errOk) {
+		t.Errorf("load %s: bad err:\nhave: %v\nwant: %v", xid, err, errOk)
+	}
+	if buf != nil {
+		t.Errorf("load %s: buf != nil", xid)
+	}
+	if serial != 0 {
+		t.Errorf("load %s: bad serial %s  ; want 0", xid, serial)
+	}
+}

go/zodb/storage/fs1/format.go

-// Copyright (C) 2017-2020  Nexedi SA and Contributors.
+// Copyright (C) 2017-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -207,7 +207,7 @@ func (txnh *TxnHeader) CloneFrom(txnh2 *TxnHeader) {
 type TxnLoadFlags int
 const (
 	LoadAll       TxnLoadFlags = 0x00 // load whole transaction header
-	LoadNoStrings              = 0x01 // do not load user/desc/ext strings
+	LoadNoStrings TxnLoadFlags = 0x01 // do not load user/desc/ext strings
 )
 
 // Load reads and decodes transaction record header @ pos.
@@ -575,7 +575,10 @@ func (dh *DataHeader) LoadBackRef(r io.ReaderAt) (backPos int64, err error) {
 	}
 
 	backPos = int64(binary.BigEndian.Uint64(dh.workMem[0:]))
-	if !(backPos == 0 || backPos >= dataValidFrom) {
+	if backPos == 0 {
+		return 0, nil // deletion
+	}
+	if backPos < dataValidFrom {
 		return 0, checkErr(r, dh, "invalid backpointer: %v", backPos)
 	}
 	if backPos + DataHeaderSize > dh.TxnPos - 8 {

go/zodb/storage/fs1/fs1tools/dump.go

-// Copyright (C) 2017  Nexedi SA and Contributors.
-//                     Kirill Smelkov <kirr@nexedi.com>
+// Copyright (C) 2017-2021  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
@@ -51,6 +51,9 @@ type Dumper interface {
 	//
 	// If dumper return io.EOF the whole dumping process finishes.
 	DumpTxn(buf *xfmt.Buffer, it *fs1.Iter) error
+
+	// DumpEndOK is called at the end of successfull dump.
+	DumpEndOK(buf *xfmt.Buffer) error
 }
 
 // Dump dumps content of a FileStorage file @ path.
@@ -101,7 +104,7 @@ func Dump(w io.Writer, path string, dir fs1.IterDir, d Dumper) (err error) {
 		err = it.NextTxn(fs1.LoadAll)
 		if err != nil {
 			if err == io.EOF {
-				err = nil
+				break
 			}
 			return err
 		}
@@ -109,7 +112,7 @@ func Dump(w io.Writer, path string, dir fs1.IterDir, d Dumper) (err error) {
 		err = d.DumpTxn(buf, it)
 		if err != nil {
 			if err == io.EOF {
-				err = nil
+				break
 			}
 			return err
 		}
@@ -119,6 +122,12 @@ func Dump(w io.Writer, path string, dir fs1.IterDir, d Dumper) (err error) {
 			return err
 		}
 	}
+
+	err = d.DumpEndOK(buf)
+	if err != nil {
+		return err
+	}
+	return nil
 }
 
 // ----------------------------------------
@@ -198,6 +207,10 @@ func (d *DumperFsDump) DumpTxn(buf *xfmt.Buffer, it *fs1.Iter) error {
 	}
 }
 
+func (d *DumperFsDump) DumpEndOK(buf *xfmt.Buffer) error {
+	return nil
+}
+
 // DumperFsDumpVerbose implements a very verbose dumper with output identical
 // to fsdump.Dumper in zodb/py originally written by Jeremy Hylton:
 //
@@ -281,6 +294,10 @@ func (d *DumperFsDumpVerbose) dumpData(buf *xfmt.Buffer, it *fs1.Iter) error {
 	return nil
 }
 
+func (d *DumperFsDumpVerbose) DumpEndOK(buf *xfmt.Buffer) error {
+	return nil
+}
+
 const dumpSummary = "dump database transactions"
 
 func dumpUsage(w io.Writer) {
@@ -383,6 +400,10 @@ func (d *DumperFsTail) DumpTxn(buf *xfmt.Buffer, it *fs1.Iter) error {
 	return nil
 }
 
+func (d *DumperFsTail) DumpEndOK(buf *xfmt.Buffer) error {
+	return nil
+}
+
 const tailSummary = "dump last few transactions of a database"
 const ntxnDefault = 10
 

go/zodb/storage/fs1/fs1tools/dump_test.go

-// Copyright (C) 2017  Nexedi SA and Contributors.
-//                     Kirill Smelkov <kirr@nexedi.com>
+// Copyright (C) 2017-2021  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
@@ -31,6 +31,10 @@ package fs1tools
 //go:generate sh -c "python2 -c 'from ZODB.FileStorage import fsdump; fsdump.main()' ../testdata/empty.fs >testdata/empty.fsdump.ok"
 //go:generate sh -c "python2 -c 'from ZODB.FileStorage.fsdump import Dumper; import sys; d = Dumper(sys.argv[1]); d.dump()' ../testdata/empty.fs >testdata/empty.fsdumpv.ok"
 
+//go:generate sh -c "python2 -m ZODB.scripts.fstail -n 1000000 ../testdata/whiteout.fs >testdata/whiteout.fstail.ok"
+//go:generate sh -c "python2 -c 'from ZODB.FileStorage import fsdump; fsdump.main()' ../testdata/whiteout.fs >testdata/whiteout.fsdump.ok"
+//go:generate sh -c "python2 -c 'from ZODB.FileStorage.fsdump import Dumper; import sys; d = Dumper(sys.argv[1]); d.dump()' ../testdata/whiteout.fs >testdata/whiteout.fsdumpv.ok"
+
 import (
 	"bytes"
 	"fmt"
@@ -50,10 +54,11 @@ func loadFile(t *testing.T, path string) string {
 	return string(data)
 }
 
-func testDump(t *testing.T, dir fs1.IterDir, d Dumper) {
-	testv := []string{"1", "empty"}
+func testDump(t *testing.T, dir fs1.IterDir, newd func() Dumper) {
+	testv := []string{"1", "empty", "whiteout"}
 	for _, tt := range testv {
 		t.Run("db=" + tt, func(t *testing.T) {
+			d   := newd()
 			buf := bytes.Buffer{}
 
 			err := Dump(&buf, fmt.Sprintf("../testdata/%s.fs", tt), dir, d)
@@ -70,9 +75,13 @@ func testDump(t *testing.T, dir fs1.IterDir, d Dumper) {
 	}
 }
 
-func TestFsDump(t *testing.T)	{ testDump(t, fs1.IterForward,  &DumperFsDump{}) }
-func TestFsDumpv(t *testing.T)	{ testDump(t, fs1.IterForward,  &DumperFsDumpVerbose{}) }
-func TestFsTail(t *testing.T)	{ testDump(t, fs1.IterBackward, &DumperFsTail{Ntxn: 1000000}) }
+func newFsDump()  Dumper { return &DumperFsDump{} }
+func newFsDumpv() Dumper { return &DumperFsDumpVerbose{} }
+func newFsTail()  Dumper { return &DumperFsTail{Ntxn: 1000000} }
+
+func TestFsDump(t *testing.T)	{ testDump(t, fs1.IterForward,  newFsDump) }
+func TestFsDumpv(t *testing.T)	{ testDump(t, fs1.IterForward,  newFsDumpv) }
+func TestFsTail(t *testing.T)	{ testDump(t, fs1.IterBackward, newFsTail) }
 
 func BenchmarkTail(b *testing.B) {
 	// FIXME small testdata/1.fs is not representative for benchmarking

go/zodb/storage/fs1/fs1tools/main.go

-// Copyright (C) 2017  Nexedi SA and Contributors.
-//                     Kirill Smelkov <kirr@nexedi.com>
+// Copyright (C) 2017-2021  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
@@ -30,7 +30,7 @@ var commands = prog.CommandRegistry{
 	{"reindex", reindexSummary, reindexUsage, reindexMain},
 	{"verify-index", verifyIdxSummary, verifyIdxUsage, verifyIdxMain},
 	// recover		(fsrecover.py)
-	// verify		(fstest.py)
+	{"verify",  verifySummary,  verifyUsage,  verifyMain},
 
 	// XXX repozo ?
 }

go/zodb/storage/fs1/fs1tools/testdata/whiteout.fsdump.ok

+Trans #00000 tid=0000000000000017 time=1900-01-01 00:00:00.000000 offset=27
+    status=' ' user='' description=''
+  data #00000 oid=0000000000000001 class=undo or abort of object creation

go/zodb/storage/fs1/fs1tools/testdata/whiteout.fsdumpv.ok

+************************************************************
+file identifier: 'FS21'
+============================================================
+offset: 4
+end pos: 77
+transaction id: 0000000000000017
+trec len: 73
+status: ' '
+user: ''
+description: ''
+len(extra): 0
+------------------------------------------------------------
+offset: 27
+oid: 0000000000000001
+revid: 0000000000000017
+previous record offset: 0
+transaction offset: 4
+len(data): 0
+backpointer: 0
+redundant trec len: 73

go/zodb/storage/fs1/fs1tools/testdata/whiteout.fstail.ok

+1900-01-01 00:00:00.000000: hash=e3e0ff6c686f2fa02266e744f6629d592d432061
+user='' description='' length=73 offset=4 (+23)
+

go/zodb/storage/fs1/fs1tools/verify.go

+// Copyright (C) 2021  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.
+
+package fs1tools
+// verify subcommand
+//
+// verification output mimics fstest from ZODB/py as originally written by Jeremy Hylton:
+// https://github.com/zopefoundation/ZODB/blob/5.6.0-35-g1fb097b41/src/ZODB/scripts/fstest.py
+
+import (
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"lab.nexedi.com/kirr/neo/go/zodb/storage/fs1"
+
+	"lab.nexedi.com/kirr/go123/prog"
+	"lab.nexedi.com/kirr/go123/xflag"
+	"lab.nexedi.com/kirr/go123/xfmt"
+)
+
+// Verify verifies content of a FileStorage file @ path.
+//
+// Only data part of the database is verified (the *.fs file).
+// Use VerifyIndexFor to verify the index part (*.fs.index).
+func Verify(w io.Writer, path string, verbose int, progress bool) (err error) {
+	// just iterate through the file and emit progress.
+	// the FileStorage driver implements all consistency checks by itself.
+	fi, err := os.Stat(path)
+	if err != nil {
+		return fmt.Errorf("verify: %s: %s", path, err)
+	}
+	fsize := fi.Size()
+
+	v := &Verifier{verbose: verbose}
+
+	// display progress updates once per tick
+	if progress {
+		tick := time.NewTicker(time.Second / 4)
+		defer tick.Stop()
+
+		xcr := ""
+		if verbose > 0 {
+			xcr = "\n"
+		}
+		v.progress = func(force bool) error {
+			if !force {
+				select {
+				case <-tick.C:
+				default:
+					return nil
+				}
+			}
+
+			_, err := fmt.Fprintf(w,
+				"\rVerified data bytes: %.1f%% (%d/%d);  #txn: %d%s",
+				100 * float64(v.donePos) / float64(fsize),
+				v.donePos, fsize,
+				v.ntxn,
+				xcr)
+			return err
+		}
+	}
+
+	return Dump(w, path, fs1.IterForward, v)
+}
+
+// Verifier implements Dumper that is used by Verify.
+type Verifier struct {
+	ntxn    int // current transaction record #
+	verbose int // >=1 (print txn)   >=2 (print objects)
+
+	// for loading data
+	dhLoading fs1.DataHeader
+
+	donePos  int64                  // done verifying till this position
+	progress func(force bool) error // called after each transaction if !nil
+}
+
+func (v *Verifier) DumperName() string {
+	return "fsverify"
+}
+
+func (v *Verifier) DumpFileHeader(buf *xfmt.Buffer, fh *fs1.FileHeader) error {
+	return nil
+}
+
+func (v *Verifier) DumpTxn(buf *xfmt.Buffer, it *fs1.Iter) error {
+	txnh := &it.Txnh
+
+	for i := 0; ; i++ {
+		err := it.NextData()
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			return err
+		}
+
+		dh := &it.Datah
+
+		// load data
+		v.dhLoading = *dh
+		dbuf, err := v.dhLoading.LoadData(it.R)
+		if err != nil {
+			return err
+		}
+
+		if v.verbose >= 2 {
+			buf .S(fmt.Sprintf("%10d: object oid 0x%s #%d\n", dh.Pos, dh.Oid, i))
+		}
+
+		dbuf.Release()
+	}
+
+	if v.verbose >= 1 {
+		buf .S(fmt.Sprintf("%10d: transaction tid 0x%s #%d \n", txnh.Pos, txnh.Tid, v.ntxn))
+	}
+
+	v.ntxn++
+
+	if v.progress != nil {
+		v.donePos = txnh.Pos + txnh.Len
+		err := v.progress(/*force=*/false)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (v *Verifier) DumpEndOK(buf *xfmt.Buffer) error {
+	if v.progress != nil {
+		err := v.progress(/*force=*/true)
+		if err != nil {
+			return err
+		}
+	}
+	if v.verbose >= 1 {
+		buf .S("no errors detected\n")
+	}
+	return nil
+}
+
+
+// ----------------------------------------
+
+const verifySummary = "verify database content"
+
+func verifyUsage(w io.Writer) {
+	fmt.Fprintf(w,
+`Usage: fs1 verify [options] <storage>
+Verify FileStorage records for consistency
+
+<storage> is a path to FileStorage
+
+  options:
+
+	-h --help       this help text.
+	-v		increase verbosity.
+	-p		display progress.
+`)
+}
+
+func verifyMain(argv []string) {
+	verbose := 0
+	var progress bool
+	flags := flag.FlagSet{Usage: func() { verifyUsage(os.Stderr) }}
+	flags.Init("", flag.ExitOnError)
+	flags.Var((*xflag.Count)(&verbose), "v", "verbosity level")
+	flags.BoolVar(&progress, "p", false, "display progress")
+	flags.Parse(argv[1:])
+
+	argv = flags.Args()
+	if len(argv) < 1 {
+		flags.Usage()
+		prog.Exit(2)
+	}
+	storPath := argv[0]
+
+	err := Verify(os.Stdout, storPath, verbose, progress)
+	if err != nil {
+		prog.Fatal(err)
+	}
+}

go/zodb/storage/fs1/py/gen-testdata

 #!/usr/bin/env python2
 # -*- coding: utf-8 -*-
-# Copyright (C) 2017-2020  Nexedi SA and Contributors.
+# Copyright (C) 2017-2021  Nexedi SA and Contributors.
 #                          Kirill Smelkov <kirr@nexedi.com>
 #
 # This program is free software: you can Use, Study, Modify and Redistribute
@@ -21,9 +21,10 @@
 """generate reference fs1 database and index for tests"""
 
 from ZODB.FileStorage import FileStorage
+from ZODB.FileStorage.FileStorage import FILESTORAGE_MAGIC, TxnHeader, DataHeader, TRANS_HDR_LEN
 from ZODB import DB
 from ZODB.Connection import TransactionMetaData
-from zodbtools.test.gen_testdata import gen_testdb, precommit, run_with_zodb3py2_compat
+from zodbtools.test.gen_testdata import gen_testdb, precommit, run_with_zodb4py2_compat
 from os import stat, remove
 from shutil import copyfile
 from golang.gcompat import qq
@@ -156,7 +157,7 @@ def main():
         vstor.store(vroot._p_oid, vroot._p_serial, '000 data 000', '', txn_stormeta)
         vstor.tpc_vote(txn_stormeta)
         # NO tpc_finish here so that status remain 'c' (voted) instead of ' ' (committed)
-    run_with_zodb3py2_compat(_)
+    run_with_zodb4py2_compat(_)
 
     st = stat(outfs)
     l  = st.st_size
@@ -174,5 +175,36 @@ def main():
     remove(voted+".lock")
 
 
+    # prepare file with whiteout (deletion of previously non-existing object)
+    whiteout = "testdata/whiteout.fs"
+    # as of 20210317 FileStorage.deleteObject verifies that object exists
+    # -> prepare magic/transaction/data records manually
+    with open(whiteout, "wb") as f:
+        oid = p64(1)
+        tid = p64(0x17)
+
+        # file header
+        f.write(FILESTORAGE_MAGIC)
+        tpos = f.tell()
+
+        # data record (see FileStorage.deleteObject)
+        dh = DataHeader(oid, tid, 0, tpos, 0, 0)
+        drec = dh.asString() + p64(0)
+
+        # emit txn header (see FileStorage.tpc_vote)
+        tlen = TRANS_HDR_LEN + 0 + 0 + 0 + len(drec) # empty u,d,e
+        th = TxnHeader(tid, tlen, ' ', 0, 0, 0)
+        th.user  = b''
+        th.descr = b''
+        th.ext   = b''
+        f.write(th.asString())
+
+        # emit data record
+        f.write(drec)
+
+        # emit txn tail
+        f.write(p64(tlen))
+
+
 if __name__ == '__main__':
     main()

go/zodb/storage/zeo/proto.go

-// Copyright (C) 2018-2020  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -54,7 +54,7 @@ type msg struct {
 type msgFlags int64
 const (
 	msgAsync  msgFlags = 1 // message does not need a reply
-	msgExcept          = 2 // exception was raised on remote side (ZEO5)
+	msgExcept msgFlags = 2 // exception was raised on remote side (ZEO5)
 )
 
 // encoding represents messages encoding.
@@ -104,7 +104,7 @@ func pktEncodeM(m msg) *pktBuf {
 	// arg
 	// it is interface{} - use shamaton/msgpack since msgp does not handle
 	// arbitrary interfaces well.
-	dataArg, err := msgpack.Encode(m.arg)
+	dataArg, err := msgpack.Marshal(m.arg)
 	if err != nil {
 		panic(err) // all our types are expected to be supported by msgpack
 	}
@@ -240,7 +240,7 @@ func pktDecodeM(pkb *pktBuf) (msg, error) {
 	if len(btail) != 0 {
 		return m, derrf(".%d: payload has extra data after message")
 	}
-	err = msgpack.Decode(b, &m.arg)
+	err = msgpack.Unmarshal(b, &m.arg)
 	if err != nil {
 		return m, derrf(".%d: arg: %s", m.msgid, err)
 	}

go/zodb/storage/zeo/zeo.go

@@ -52,6 +52,9 @@ type zeo struct {
 	// becomes ready when serve loop finishes
 	serveWG sync.WaitGroup
 
+	closeOnce sync.Once
+	closed    chan struct{} // ready when Closed
+
 	url string // we were opened via this
 }
 
@@ -174,7 +177,13 @@ func (z *zeo) invalidateTransaction(arg interface{}) (err error) {
 
 	// at0 is initialized - ok to send current event if it goes > at0
 	if tid > z.at0 {
-		z.watchq <- event
+		select {
+		case <-z.closed:
+			// closed - client does not read watchq anymore
+
+		case z.watchq <- event:
+			// ok
+		}
 	}
 	return nil
 }
@@ -188,7 +197,13 @@ func (z *zeo) flushEventq0() {
 	if z.watchq != nil {
 		for _, e := range z.eventq0 {
 			if e.Tid > z.at0 {
-				z.watchq <- e
+				select {
+				case <-z.closed:
+					// closed - client does not read watchq anymore
+
+				case z.watchq <- e:
+					// ok
+				}
 			}
 		}
 	}
@@ -264,7 +279,7 @@ func (r rpc) call(ctx context.Context, argv ...interface{}) (interface{}, error)
 // excError returns error corresponding to an exception.
 //
 // well-known exceptions are mapped to corresponding well-known errors - e.g.
-// POSKeyError -> zodb.NoObjectError, and rest are returned wrapper into rpcExcept.
+// POSKeyError -> zodb.NoObjectError, and rest are returned wrapped into rpcExcept.
 func (r rpc) excError(exc string, argv tuple) error {
 	// translate well-known exceptions
 	switch exc {
@@ -438,7 +453,7 @@ func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb
 	}()
 
 
-	z := &zeo{link: zlink, watchq: opt.Watchq, url: url}
+	z := &zeo{link: zlink, watchq: opt.Watchq, closed: make(chan struct{}), url: url}
 
 	// start serve loop on the link
 	z.serveWG.Add(1)
@@ -456,14 +471,18 @@ func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb
 		// close .watchq after serve is over
 		z.at0Mu.Lock()
 		defer z.at0Mu.Unlock()
-		if z.at0Initialized {
-			z.flushEventq0()
-		}
 		if z.watchq != nil {
-			if err != nil {
-				z.watchq <- &zodb.EventError{Err: err}
+			if err != nil && /* already flushed .eventq0 */z.at0Initialized {
+				select {
+				case <-z.closed:
+					// closed - client does not read watchq anymore
+
+				case z.watchq <- &zodb.EventError{Err: err}:
+					// ok
+				}
 			}
 			close(z.watchq)
+			z.watchq = nil // prevent flushEventq0 to send to closed chan
 		}
 	}()
 
@@ -496,13 +515,16 @@ func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb
 	// "invalidateTransaction" server notification.
 	//
 	// filter-out first < at0 messages for this reason.
-	z.at0Mu.Lock()
-	z.at0 = lastTid
-	z.at0Initialized = true
-	z.flushEventq0()
-	z.at0Mu.Unlock()
-
-
+	//
+	// do this in separate task not to deadlock in watchq<- : we did not
+	// yet returned z to caller and so noone might be yet reading from watchq.
+	go func() {
+		z.at0Mu.Lock()
+		z.at0 = lastTid
+		z.at0Initialized = true
+		z.flushEventq0()
+		z.at0Mu.Unlock()
+	}()
 
 	//call('get_info') -> {}str->str, ex	// XXX can be omitted
 /*
@@ -520,11 +542,15 @@ func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb
   'supports_record_iternext': True})
 */
 
-	return z, z.at0, nil
+	return z, lastTid, nil
 }
 
 func (z *zeo) Close() error {
-	err := z.link.Close()
+	var err error
+	z.closeOnce.Do(func() {
+		close(z.closed)
+		err = z.link.Close()
+	})
 	z.serveWG.Wait()
 	return err
 }

go/zodb/storage/zeo/zeo_test.go

@@ -153,7 +153,7 @@ func (z *ZEOPySrv) Encoding() encoding {
 // ----------------
 
 // tOptions represents options for testing.
-// XXX dup in NEO
+// TODO -> xtesting
 type tOptions struct {
 	Preload string // preload database with data from this location
 }

go/zodb/wks/wks.go

-// Copyright (C) 2017  Nexedi SA and Contributors.
-//                     Kirill Smelkov <kirr@nexedi.com>
+// Copyright (C) 2017-2021  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
@@ -31,4 +31,5 @@ import (
 	_ "lab.nexedi.com/kirr/neo/go/zodb/storage/fs1"
 	_ "lab.nexedi.com/kirr/neo/go/zodb/storage/zeo"
 	_ "lab.nexedi.com/kirr/neo/go/neo"
+	_ "lab.nexedi.com/kirr/neo/go/zodb/storage/demo"
 )

go/zodb/zodb.go

-// Copyright (C) 2016-2020  Nexedi SA and Contributors.
+// Copyright (C) 2016-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -275,7 +275,7 @@ type NoObjectError struct {
 	Oid Oid
 }
 
-func (e NoObjectError) Error() string {
+func (e *NoObjectError) Error() string {
 	return fmt.Sprintf("%s: no such object", e.Oid)
 }
 

go/zodb/zodbtools/help.go

@@ -41,9 +41,10 @@ and using path to that file with zconfig:// schema:
 
 There are also following simpler ways:
 
-- neo://<master>/<db>   for a NEO database	XXX + neos:// ?
-- zeo://<host>:<port>   for a ZEO database
-- /path/to/file         for a FileStorage database
+- neo://<master>/<db>         for a NEO database	XXX + neos:// ?
+- zeo://<host>:<port>         for a ZEO database
+- /path/to/file               for a FileStorage database
+- demo:(zurl_base)/(zurl_δ)   for a DemoStorage overlay
 
 Please see zodburi documentation for full details:
 

go/zodb/δtail.go

-// Copyright (C) 2018-2020  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -202,46 +202,35 @@ func (δtail *ΔTail) ForgetPast(revCut Tid) {
 // XXX -> RevAt ?
 // LastRevOf tries to return what was the last revision that changed id as of at database state.
 //
-// Depending on current information in δtail it returns either exact result, or
-// an upper-bound estimate for the last id revision, assuming id was changed ≤ at:
+// it must be called with the following condition:
 //
-// 1) if δtail does not cover at, at is returned:
+//	tail ≤ at ≤ head
 //
-//	# at ∉ [min(rev ∈ δtail), max(rev ∈ δtail)]
-//	LastRevOf(id, at) = at
+// Depending on current information in δtail it returns either exact result, or
+// an upper-bound estimate for the last id revision:
 //
-// 2) if δtail has an entry corresponding to id change, it gives exactly the
+// 1) if δtail has an entry corresponding to id change, it gives exactly the
 // last revision that changed id:
 //
-//	# at ∈ [min(rev ∈ δtail), max(rev ∈ δtail)]
 //	# ∃ rev ∈ δtail: rev changed id && rev ≤ at
-//	LastRevOf(id, at) = max(rev: rev changed id && rev ≤ at)
+//	LastRevOf(id, at) = max(rev: rev changed id && rev ≤ at), true
 //
-// 3) if δtail does not contain appropriate record with id - it returns δtail's
+// 2) if δtail does not contain appropriate record with id - it returns δtail's
 // lower bound as the estimate for the upper bound of the last id revision:
 //
-//	# at ∈ [min(rev ∈ δtail), max(rev ∈ δtail)]
 //	# ∄ rev ∈ δtail: rev changed id && rev ≤ at
-//	LastRevOf(id, at) = min(rev ∈ δtail)
+//	LastRevOf(id, at) = δtail.tail, false
 //
 // On return exact indicates whether returned revision is exactly the last
 // revision of id, or only an upper-bound estimate of it.
 func (δtail *ΔTail) LastRevOf(id Oid, at Tid) (_ Tid, exact bool) {
-	// check if we have no coverage at all
-	l := len(δtail.tailv)
-	if l == 0 {
-		return at, false
-	}
-	revMin := δtail.tailv[0].Rev
-	revMax := δtail.tailv[l-1].Rev
-	if !(revMin <= at && at <= revMax) {
-		return at, false
-	}
+	if !(δtail.tail <= at && at <= δtail.head) {
+		panic(fmt.Sprintf("at out of bounds: at: @%s,  (tail, head] = (@%s, @%s]", at, δtail.tail, δtail.head))
+        }
 
-	// we have the coverage
 	rev, ok := δtail.lastRevOf[id]
 	if !ok {
-		return δtail.tailv[0].Rev, false
+		return δtail.tail, false
 	}
 
 	if rev <= at {
@@ -250,7 +239,7 @@ func (δtail *ΔTail) LastRevOf(id Oid, at Tid) (_ Tid, exact bool) {
 
 	// what's in index is after at - scan tailv back to find appropriate entry
 	// XXX linear scan - see .lastRevOf comment.
-	for i := l - 1; i >= 0; i-- {
+	for i := len(δtail.tailv) - 1; i >= 0; i-- {
 		δ := δtail.tailv[i]
 		if δ.Rev > at {
 			continue
@@ -264,5 +253,5 @@ func (δtail *ΔTail) LastRevOf(id Oid, at Tid) (_ Tid, exact bool) {
 	}
 
 	// nothing found
-	return δtail.tailv[0].Rev, false
+	return δtail.tail, false
 }

go/zodb/δtail_test.go

-// Copyright (C) 2018-2019  Nexedi SA and Contributors.
+// Copyright (C) 2018-2021  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -174,14 +174,13 @@ func TestΔTail(t *testing.T) {
 	δtail = NewΔTail(3)
 
 	δCheck(3,3)
-	δCheckLastUP(4, 12, 12)	// δtail = ø
+	δCheckLastUP(3, 3, 3)	// δtail = ø
 
 	δAppend(R(10, 3,5))
 	δCheck(3,10, R(10, 3,5))
 
-	δCheckLastUP(3,  2,  2)	// at < δtail
-	δCheckLastUP(3, 12, 12)	// at > δtail
-	δCheckLastUP(4, 10, 10)	// id ∉ δtail
+	δCheckLastUP(3,  9, 3)  // id ∈ δtail, but has no entry with rev ≤ at
+	δCheckLastUP(4, 10, 3)	// id ∉ δtail
 
 	δAppend(R(11, 7))
 	δCheck(3,11, R(10, 3,5), R(11, 7))
@@ -192,7 +191,7 @@ func TestΔTail(t *testing.T) {
 	δAppend(R(14, 3,8))
 	δCheck(3,14, R(10, 3,5), R(11, 7), R(12, 7), R(14, 3,8))
 
-	δCheckLastUP(8, 12, 10) // id ∈ δtail, but has no entry with rev ≤ at
+	δCheckLastUP(8, 12, 3)  // id ∈ δtail, but has no entry with rev ≤ at
 
 	δtail.ForgetPast(9)
 	δCheck(9,14, R(10, 3,5), R(11, 7), R(12, 7), R(14, 3,8))

neo/client/zodburi.py

@@ -84,6 +84,7 @@ def _resolve_uri(uri):
         if scheme != "neos":
             raise ValueError("invalid uri: %s : credentials can be specified only with neos:// scheme" % uri)
         # ca=ca.crt;cert=my.crt;key=my.key
+        cred = cred.replace(';', '&') # ; is no longer in default separators set bugs.python.org/issue42967
         for k, v in OrderedDict(parse_qsl(cred)).items():
             if k not in _credopts:
                 raise ValueError("invalid uri: %s : unexpected credential %s" % (uri, k))

neo/tests/ca.crt

 -----BEGIN CERTIFICATE-----
-MIIC7TCCAdWgAwIBAgIJAL8e44sA7PDMMA0GCSqGSIb3DQEBCwUAMA0xCzAJBgNV
-BAMMAkNBMB4XDTE1MDkzMDEzNTQzMFoXDTIxMDMyMjEzNTQzMFowDTELMAkGA1UE
-AwwCQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCgT7DVKM4ViQt3
-B0oJC4RGi10dNfpNZZpgA5iC2UJ1u6AqqCf0PCQkrmuIzW3l1TenlOiLNdVASkkT
-wf1lekIgg4tR8/22oGTAnfY6R9r1C6jAMV72v1sffz8D6qfkMPzKchJt55zywdhm
-KscUsMGzXPGIeKrG20m83dSIO4RmCmq/f4BcuWJu6Kkq4n9Wc2IsvpKk+lqEUxI/
-QoqdT6OvMXooGs3t892uvKDu++muBj2Y/yyaXt1tCCjDFsRMLWl3Skks+4PeMCZ4
-wugyXEBk3d5Yzdv5NsFzFBjAuRCGxJXEOEcfHj4Xj9qTCErZ1jKzgnuxJCtgdqRC
-r4beX1U3AgMBAAGjUDBOMB0GA1UdDgQWBBSFY/jKvo0iSTEzzOIcZZUZCT8JfTAf
-BgNVHSMEGDAWgBSFY/jKvo0iSTEzzOIcZZUZCT8JfTAMBgNVHRMEBTADAQH/MA0G
-CSqGSIb3DQEBCwUAA4IBAQAB0LKDAuhodpyNVwEE9Yl+Q/IiPEPCaix6URJnRn1O
-gQnXuZLo1xtJh6wJh1faG1/qNCFMxWEJ+0VkJ7r6v38cNXfYG9OcmD0S6YnNjSuO
-VliAtqVVtj8MppJ4vMatLrNi4cvyYucebtNyBCzSIAi+6bkkHeaVgi1EtxXvq+AS
-iZp3gl84oXv/gV7Bz4SXmVpFJnhsDMoQZG2KAULAgfZ2Am2I+ffG90cD/oEnS/3O
-k3btqTvgIO8MWt8PY3sUOhJEoJYKnC9DppmhOhUTn4zzIIDSluKEOBHZiFb9AcmF
-PvzL+8xiORCdUe1d6ANQQlUd0MM810BXZFYEXFbgKg8o
+MIIC+zCCAeOgAwIBAgIUN2aMGxxsLX4IyuSR2DRfzbTumrAwDQYJKoZIhvcNAQEL
+BQAwDTELMAkGA1UEAwwCQ0EwHhcNMjEwMzIyMTU1MzI5WhcNMjYwOTEyMTU1MzI5
+WjANMQswCQYDVQQDDAJDQTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
+AMVipTeJN6FEpdF6JJkkVw7vPR14pXCMehSWxBMrmICVAHct4fWz4sI/lp3VTiFl
+HnEBOhehLsKfZyosGIMpJ0zI2GsQopkHa5ZPf/cILrQ3ybu7ZsQITAyYKWqhMVdt
+d+YRetfkE874gUHqTo+WSBpCbRvuYK4XhBEmrMiwIHV7nwuIaKMdG6FxNGn8du8R
+Ti96VduJGdjTFIRri2jNP47VKidioPkxecA/48yrDdewn/cxsU/yNCO9Uii/dTyS
+Ro/Jqxql6Leb7xaZlODS5QQAGWjAI6w0BCobwYaUuJOZ1HSUL6A42rnMqI6gGjCY
+Dju313RlE5jOilWQhodmawECAwEAAaNTMFEwHQYDVR0OBBYEFBfzeqZl9fHZR2bH
+rb+T6+O6mu1aMB8GA1UdIwQYMBaAFBfzeqZl9fHZR2bHrb+T6+O6mu1aMA8GA1Ud
+EwEB/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBAK6+04BOOP/PF6BR6FaZjLHN
+qhcSBfG4Y8RyLtERwGTQW68PwCU2TloJgpoY8updT0V9EsxnkWjsU/Wgnl8pdHjp
+GMAivelTcDoYH4qn8E6mDEKE47Be4fE0FdGvU9QU7UBWG+Uf8pxgnFJcGou5grLG
+RXTSNCSrBnJG9tmSPA/aw/Q9MVfXFuB7K8aPJ/d2Z/8v/QFdBK25JH8I+ErFU3ys
+ufn58Jqsy4dp8R4qYVP1TKPfGsZ+RZeIIJ4cDIik/vbv9+Ymi+m8LZ66kUNq1d4X
+tGbPjcNt/qFQ/FG7T6ZUPe9vRpYAhIHGBYZFs59LJcQsH1DzSvei3141Gs7TYFg=
 -----END CERTIFICATE-----

neo/tests/node.crt

 -----BEGIN CERTIFICATE-----
-MIICkDCCAXgCAQEwDQYJKoZIhvcNAQELBQAwDTELMAkGA1UEAwwCQ0EwHhcNMTUw
-OTMwMTM1NDMwWhcNMjEwMzIyMTM1NDMwWjAPMQ0wCwYDVQQDDARub2RlMIIBIjAN
+MIICkDCCAXgCAQEwDQYJKoZIhvcNAQELBQAwDTELMAkGA1UEAwwCQ0EwHhcNMjEw
+MzIyMTU1MzI5WhcNMjYwOTEyMTU1MzI5WjAPMQ0wCwYDVQQDDARub2RlMIIBIjAN
 BgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAuJ+ClJyjhJOJdGUyqHn79opMLP3m
 1g27uBWKT+OGd4FcreVoRDxPVuxZxMtDCZcBfUHVvOoSlS06khwSxViEe1hxwHRa
 n2qMWlwvaWeNY0CFH5V+DI4XSNojgny85Lb5jB69FuPcrHnwxLk2OFntrXEeNbEa
 d7QSoNbPajbJIp5BS/WR9iu5Z5JYdumLWjTvOU+eZc4iA6Wa2kdDtbqkGi4wOJ1L
 /ggATL+p+QFcubVptztPT8vq7gvDdGJgXLJ2lPHV0V/sdJB1FB4mSJEDDjSm2Hpp
 qPVJSO1GrAy5Ld+0SnXIZZejhIUJIumocY08r+vzDSQ/8NnqXR4Odz1TWwIDAQAB
-MA0GCSqGSIb3DQEBCwUAA4IBAQBlYkkInDDDcgnNRdUmzxwejs1PmEehZ3H5FkMp
-TsmpoVC+oqM+QywMu8UJRtCjXnnJdAUbVYuZ1Tjm7qvFIhN+5OlIVxJ+8WcmZPSe
-lj0N7Dv2nE1diTDS+qPZVPZ0demo1LafRmPomPWiM/CQRlMPxXnimuiYOROhWGn6
-jsyoOwquMkAc6Ub++l4OCxLAP0eTgJFkivmqpaYZXG4o7zFvcQ3rQ66rQrMl69sR
-8/MVqbT5Sq1CEJbepP4GaFfa5l3CVy7WH2MhCV1/9mNwcXafkTgx3q2HsPon4Dze
-kNwiguNAM4L/j4dbIwz+CIVWcgpBCrfv2JYu+jGlRpxIDeWR
+MA0GCSqGSIb3DQEBCwUAA4IBAQAvtJztgZ8TuY5Q52Lv4hE2WX3/IZQPYc9cGQqv
+sI0xKQEWD4hVHAmxL7bLFOUovzFqNMY0pYBGVJ53nypsKNorKbiepSMPosfcbk9o
+U8xF9YDJO+yS6V2lWFi6iKe4WX2t+L2j5x+q4dptJoIvsM777xGatri59q8UK3ne
+YTINui8+tsOIGEpcXMHMI/k3RvBUS3Mwy0pX6rsi8/YsUVUShpPDjuAqQIYxU4gP
+qAIWrhXNYwOGuMQ1V+j83R5lqDoHrVauWznc0n0qg/EJArhwkLGfjQoqrogCQOVp
+iMEV6YDLC8EMqZfYVUT0i2+/bS737Z2lcdgz2essW45nk7+H
 -----END CERTIFICATE-----