go/zodb/demo: New package that provides base+δ storages overlay

For Load demo.Storage implementation is similar to DemoStorage in
ZODB/py with fixes "cherry-picked" from:

- https://github.com/zopefoundation/ZODB/issues/318
  (DemoStorage does not take whiteouts into account -> leading to data corruption)

- https://github.com/zopefoundation/ZODB/pull/323
  (loadAt + fix for the above issue)

For safety demo.Storage - contrary to DemoStorage/py - actually verifies
that for demo=base+δ δ comes strictly after base and that base remains
unchanged.

URI schema follows XRI Cross-references approach and is

	demo:(zurl_base)/(zurl_δ)

https://en.wikipedia.org/wiki/Extensible_Resource_Identifier provides
some related details and examples.

For ZODB/py corresponding pull-request for zodburi to support demo: URI
scheme has been made here: https://github.com/Pylons/zodburi/pull/29 .

Tests need:

- recent zodbtools with zodbrestore:

  https://lab.nexedi.com/nexedi/zodbtools/blob/129afa67/zodbtools/zodbrestore.py
  nexedi/zodbtools!19

- ZODB with support for DemoStorage.deleteObject
  https://github.com/zopefoundation/ZODB/pull/341

On Go side demo storage is needed for wendelin.core 2 because ERP5 uses
DemoStorage to run tests.

go/internal/xtesting/xtesting.go

@@ -29,11 +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"
 )
 
@@ -134,6 +136,33 @@ func ZPyCommitRaw(zurl string, at zodb.Tid, objv ...ZRawObject) (_ zodb.Tid, err
 	return tid, nil
 }
 
+// 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("python2", "-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 ----
 

go/zodb/storage/demo/demo.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 provides overlayed storage, similar to DemoStorage in ZODB/py.
+//
+// Storage combines base and δ storages as if δ transactional log is logically
+// appended to base.
+package demo
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/url"
+	"regexp"
+	"sync"
+
+	"lab.nexedi.com/kirr/go123/mem"
+	"lab.nexedi.com/kirr/go123/xerr"
+	"lab.nexedi.com/kirr/go123/xsync"
+	"lab.nexedi.com/kirr/neo/go/internal/task"
+	"lab.nexedi.com/kirr/neo/go/zodb"
+)
+
+// Storage combines base and δ storages as if δ transactional log is logically
+// appended to base.
+//
+// Base must remain unmodified, and all δ transactions must be with TIDs
+// strictly after base.Head .
+type Storage struct {
+	base zodb.IStorageDriver
+	δ    zodb.IStorageDriver
+
+	baseAt0    zodb.Tid
+	baseWatchq <-chan zodb.Event
+	δWatchq    <-chan zodb.Event // nil if demo is opened with watchq=nil
+	δWatchq0   <-chan zodb.Event // buffer for δ events queued while δ was initially verified
+	watchq     chan<- zodb.Event // user requested to deliver events here
+
+	watchWG     *xsync.WorkGroup
+	watchCancel func()
+
+	downOnce sync.Once
+	down     chan struct{} // ready when storage is down
+	downErr  error         // reason for shutdown
+
+	closeOnce sync.Once
+	closed    chan struct{} // ready when storage is Closed
+}
+
+// baseMutatedError is reported when Storage.base is detected to change.
+type baseMutatedError struct {
+	baseAt0  zodb.Tid
+	baseHead zodb.Tid
+}
+
+func (e *baseMutatedError) Error() string {
+	return fmt.Sprintf("base.head mutated from @%s to @%s", e.baseAt0, e.baseHead)
+}
+
+// baseErrorEvent is reported on error event detected on Storage.base .
+type baseErrorEvent struct {
+	err error
+}
+
+func (e *baseErrorEvent) Error() string {
+	return fmt.Sprintf("base: error event: %s", e.err)
+}
+
+func (e *baseErrorEvent) Cause() error  { return e.err }
+func (e *baseErrorEvent) Unwrap() error { return e.err }
+
+
+// watcher task detects base mutation and proxies δ events to user watchq.
+func (d *Storage) watcher(ctx context.Context) error {
+	if d.watchq != nil {
+		defer close(d.watchq)
+	}
+
+	for {
+		var δWatchq <-chan zodb.Event
+		if len(d.δWatchq0) != 0 {
+			δWatchq = d.δWatchq0
+		} else {
+			δWatchq = d.δWatchq
+		}
+
+		select {
+		// Close requests to stop watching
+		case <-ctx.Done():
+			return ctx.Err()
+
+		// event on base -> base mutated/error + shutdown
+		case event, ok := <-d.baseWatchq:
+			if !ok {
+				// base closed
+				d.baseWatchq = nil
+				continue
+			}
+
+			var edown error
+			switch event := event.(type) {
+			case *zodb.EventCommit:
+				edown = &baseMutatedError{d.baseAt0, event.Tid}
+
+			case *zodb.EventError:
+				edown = &baseErrorEvent{event.Err}
+
+			default:
+				edown = fmt.Errorf("base: unexpected event %T", event)
+			}
+
+			d.shutdown(edown)
+			ev := &zodb.EventError{&zodb.OpError{URL: d.URL(), Op: "watcher", Err: edown}}
+			if d.watchq != nil {
+				select {
+				case <-d.closed:
+					// wakeup to return edown
+
+				case d.watchq <- ev:
+					// ok
+				}
+			}
+			return edown
+
+		// event on δ -> proxy to user
+		case event, ok := <-δWatchq:
+			if !ok {
+				// δ closed
+				d.δWatchq = nil
+				continue
+			}
+
+			select {
+			case <-d.closed:
+				return nil
+
+			case d.watchq <- event: // !nil because d.δWatchq != nil
+				// ok
+			}
+		}
+	}
+}
+
+// shutdown marks Storage as no longer operational due to reason.
+func (d *Storage) shutdown(reason error) {
+	d.downOnce.Do(func() {
+		d.downErr = reason
+		close(d.down)
+	})
+}
+
+var errClosed = errors.New("storage is closed")
+
+// Close implements zodb.IStorageDriver .
+func (d *Storage) Close() (err error) {
+	defer func() {
+		if err != nil {
+			err = &zodb.OpError{URL: d.URL(), Op: "close", Err: err}
+		}
+	}()
+
+	d.shutdown(errClosed)
+	d.closeOnce.Do(func() {
+		close(d.closed)
+	})
+	errδ    := d.δ.Close()
+	errBase := d.base.Close()
+
+	// cancel watcher; don't propagate its error to Close - watcher error
+	// goes to watchq and op errors.
+	d.watchCancel()
+	_ = d.watchWG.Wait()
+
+	return xerr.Merge(errδ, errBase)
+}
+
+// Sync implements zodb.IStorageDriver .
+func (d *Storage) Sync(ctx context.Context) (_ zodb.Tid, err error) {
+	defer func() {
+		if err != nil {
+			err = &zodb.OpError{URL: d.URL(), Op: "sync", Err: err}
+		}
+	}()
+	if ready(d.down) {
+		return zodb.InvalidTid, d.downErr
+	}
+
+	var head zodb.Tid
+	wg := xsync.NewWorkGroup(ctx)
+	wg.Go(func(ctx context.Context) error {
+		h, err := d.δ.Sync(ctx)
+		head = h
+		return err
+	})
+	wg.Go(func(ctx context.Context) error {
+		baseHead, err := d.base.Sync(ctx)
+		if err != nil {
+			return err
+		}
+		if baseHead != d.baseAt0 {
+			return &baseMutatedError{d.baseAt0, baseHead}
+		}
+		return nil
+	})
+
+	err = wg.Wait()
+	if err != nil {
+		return zodb.InvalidTid, err
+	}
+
+	// δ is just created database
+	if head == 0 {
+		head = d.baseAt0
+	}
+
+	return head, nil
+}
+
+// Load implements zodb.IStorageDriver .
+func (d *Storage) Load(ctx context.Context, xid zodb.Xid) (_ *mem.Buf, _ zodb.Tid, err error) {
+	defer func() {
+		if err != nil {
+			err = &zodb.OpError{URL: d.URL(), Op: "load", Args: xid, Err: err}
+		}
+	}()
+	if ready(d.down) {
+		return nil, zodb.InvalidTid, d.downErr
+	}
+
+	var eNoData   *zodb.NoDataError
+	var eNoObject *zodb.NoObjectError
+
+	inδ := false
+	if xid.At > d.baseAt0 {
+		data, serial, err := d.δ.Load(ctx, xid)
+		if err == nil {
+			// object data is present in δ
+			return data, serial, nil
+		}
+
+		useBase := false
+		switch {
+		case errors.As(err, &eNoData):
+			if eNoData.DeletedAt != 0 {
+				// object deleted in δ -> whiteout
+				return data, serial, eNoData
+			} else {
+				// object present in δ but not yet created as of xid.at
+				useBase = true
+				inδ = true
+			}
+
+		case errors.As(err, &eNoObject):
+			// object not created in δ
+			useBase = true
+		}
+
+		if !useBase {
+			return data, serial, err
+		}
+	}
+
+	// cap xid.At to .baseAt0 (we convert it back on error return, and
+	// it makes Load more robust wrt simultaneous base mutation).
+	xidBase := xid
+	if xid.At > d.baseAt0 {
+		xidBase.At = d.baseAt0
+	}
+
+	data, serial, err := d.base.Load(ctx, xidBase)
+	if err == nil {
+		return data, serial, nil
+	}
+
+	switch {
+	case errors.As(err, &eNoData):
+		err = eNoData
+	case errors.As(err, &eNoObject):
+		if !inδ {
+			err = eNoObject
+		} else {
+			// object is present in δ
+			err = &zodb.NoDataError{Oid: xid.Oid, DeletedAt: 0}
+		}
+	}
+
+	return data, serial, err
+}
+
+// Iterator implements zodb.IStorageDriver .
+func (d *Storage) Iterate(ctx context.Context, tidMin, tidMax zodb.Tid) zodb.ITxnIterator {
+	panic("TODO")
+}
+
+// URL implements zodb.IStorageDriver .
+func (d *Storage) URL() string {
+	return "demo:(" + d.base.URL() + ")/(" + d.δ.URL() + ")"
+}
+
+var demoRe = regexp.MustCompile(`^\((.*)\)/\((.*)\)$`)
+
+func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb.IStorageDriver, _ zodb.Tid, err error) {
+	// demo:(base_zurl)/(δ_zurl)
+	defer task.Runningf(&ctx, "demo: open %s", u)(&err)
+
+	if !(u.Opaque != "" && u.RawQuery == "" && u.Fragment == "") {
+		// opaque != "" makes sure user,host,path are empty
+		return nil, zodb.InvalidTid, fmt.Errorf("invalid url")
+	}
+
+	argv := demoRe.FindStringSubmatch(u.Opaque)
+	if argv == nil {
+		return nil, zodb.InvalidTid, fmt.Errorf("invalid url")
+	}
+
+	baseZURL := argv[1]
+	δZURL    := argv[2]
+
+	// open base - always readonly
+	baseWatchq := make(chan zodb.Event)
+	base, baseAt0, err := zodb.OpenDriver(ctx, baseZURL, &zodb.DriverOptions{
+		ReadOnly: true,
+		Watchq:   baseWatchq,
+	})
+	if err != nil {
+		return nil, zodb.InvalidTid, err
+	}
+	defer func() {
+		if err != nil {
+			__ := base.Close()
+			err = xerr.Merge(err, __)
+		}
+	}()
+
+	// open δ    - as requested  but with events going through us
+	δopt := *opt
+	var δWatchq chan zodb.Event
+	if δopt.Watchq != nil {
+		δWatchq = make(chan zodb.Event)
+		δopt.Watchq = δWatchq
+	}
+	δ, δAt0, err := zodb.OpenDriver(ctx, δZURL, &δopt)
+	if err != nil {
+		return nil, zodb.InvalidTid, err
+	}
+	defer func() {
+		if err != nil {
+			__ := δ.Close()
+			err = xerr.Merge(err, __)
+		}
+	}()
+
+
+	// verify that either
+	// - δ is all empty (just created), or
+	// - all δ transactions come strictly after base.Head .
+	at0 := baseAt0
+	var δEventq0 []zodb.Event
+	if δAt0 != 0 {
+		if δAt0 < baseAt0 {
+			return nil, zodb.InvalidTid, fmt.Errorf("base is ahead of δ: base.head=%s  δ.head=%s", baseAt0, δAt0)
+		}
+
+		// read and queue data from δWatchq while we verify δ (not to deadlock δ driver)
+		δq0Stop := make(chan struct{}) // reader <- main  "stop"
+		δq0Done := make(chan struct{}) // reader -> main  "done"
+		go func() {
+			defer close(δq0Done)
+
+			for {
+				select {
+				case <-δq0Stop:
+					return
+
+				case δevent, ok := <-δWatchq:
+					if !ok {
+						return
+					}
+					δEventq0 = append(δEventq0, δevent)
+				}
+			}
+		}()
+
+		// verify δ
+		iδ := δ.Iterate(ctx, 0, baseAt0)
+		δtxni, _, err := iδ.NextTxn(ctx)
+		switch {
+		case err == io.EOF:
+			err = nil // ok - nothing in δ
+
+		case err == nil:
+			// there is a δ transaction ∈ [δAt0, baseAt0)
+			err = fmt.Errorf("base overlaps with δ: base.head=%s  δ.tail=%s", baseAt0, δtxni.Tid)
+		}
+		// TODO iδ.Close()
+
+		close(δq0Stop)
+		<-δq0Done
+		if err != nil {
+			return nil, zodb.InvalidTid, err
+		}
+
+		at0 = δAt0
+	}
+
+	// requeue δWatchq0 <- δEventq0
+	var δWatchq0 chan zodb.Event
+	if l := len(δEventq0); l != 0 {
+		δWatchq0 = make(chan zodb.Event, l)
+		for _, ev := range δEventq0 {
+			δWatchq0 <- ev
+		}
+	}
+
+	d := &Storage{
+		base: base,
+		δ   : δ,
+
+		baseAt0:    baseAt0,
+		baseWatchq: baseWatchq,
+		δWatchq:    δWatchq,
+		δWatchq0:   δWatchq0,
+		watchq:     opt.Watchq,
+
+		down:   make(chan struct{}),
+		closed: make(chan struct{}),
+	}
+
+	// spawn watcher to listen on baseWatchq and shutdown storage if base changes.
+	ctx, cancel := context.WithCancel(context.Background())
+	d.watchWG = xsync.NewWorkGroup(ctx)
+	d.watchCancel = cancel
+	d.watchWG.Go(d.watcher)
+
+	return d, at0, nil
+}
+
+func init() {
+	zodb.RegisterDriver("demo", openByURL)
+}
+
+
+// ---- misc ----
+
+// ready returns whether c is ready.
+func ready(c <-chan struct{}) bool {
+	select {
+	case <-c:
+		return true
+	default:
+		return false
+	}
+}

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/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
@@ -30,4 +30,5 @@ package wks
 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/zodb/storage/demo"
 )

go/zodb/zodbtools/help.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
@@ -41,9 +41,10 @@ and using path to that file with zconfig:// schema:
 
 There are also following simpler ways:
 
-- neo://<db>@<master>   for a NEO database
-- zeo://<host>:<port>   for a ZEO database
-- /path/to/file         for a FileStorage database
+- neo://<db>@<master>         for a NEO database
+- 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:
 

[Kirill Smelkov]

On Go side demo storage is needed for wendelin.core 2 because ERP5 uses DemoStorage to run tests.

/cc @romain