go/zodb: Require drivers to provide at₀ on open

and to deliver to watchq only and all events in (at₀, +∞] range.

This continues 4d2c8b1d (go/zodb: Require drivers to provide
notifications for database change events) and makes initialization
semantics to process invalidations easier for both application-level ZODB
layer, and for low-level users that work with drivers directly:

- there is no possibility that an event will come from watchq with tid <
  current user head.

- there is no possibility that watcher will start delivering events
  after at₀, but not immediately after it, i.e. users can rely that they
  won't loose an event.

This correctness invariants should be easy to provide in drivers.

go/zodb/open.go

@@ -47,12 +47,18 @@ type DriverOptions struct {
 	//
 	// 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.
+	//
 	// TODO extend watchq to also receive errors from watcher.
 	Watchq chan<- CommitEvent
 }
 
 // DriverOpener is a function to open a storage driver.
-type DriverOpener func (ctx context.Context, u *url.URL, opt *DriverOptions) (IStorageDriver, error)
+//
+// at₀ gives database state at open time. The driver will send to Watchq (see
+// DriverOptions) only and all events in (at₀, +∞] range.
+type DriverOpener func (ctx context.Context, u *url.URL, opt *DriverOptions) (_ IStorageDriver, at0 Tid, _ error)
 
 // {} scheme -> DriverOpener
 var driverRegistry = map[string]DriverOpener{}
@@ -98,7 +104,7 @@ func OpenStorage(ctx context.Context, storageURL string, opt *OpenOptions) (ISto
 		Watchq:   nil,          // TODO use watchq to implement high-level watching
 	}
 
-	storDriver, err := opener(ctx, u, drvOpt)
+	storDriver, _, err := opener(ctx, u, drvOpt) // TODO use at0 to initialize watcher δtail
 	if err != nil {
 		return nil, err
 	}

go/zodb/storage/fs1/filestorage.go

@@ -717,10 +717,10 @@ func (fs *FileStorage) Close() error {
 }
 
 // Open opens FileStorage @path.
-func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileStorage, err error) {
+func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileStorage, at0 zodb.Tid, err error) {
 	// TODO read-write support
 	if !opt.ReadOnly {
-		return nil, fmt.Errorf("fs1: %s: TODO write mode not implemented", path)
+		return nil, zodb.InvalidTid, fmt.Errorf("fs1: %s: TODO write mode not implemented", path)
 	}
 
 	fs := &FileStorage{
@@ -730,7 +730,7 @@ func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileSto
 
 	f, err := os.Open(path)
 	if err != nil {
-		return nil, err
+		return nil, zodb.InvalidTid, err
 	}
 	fs.file = f
 	defer func() {
@@ -745,7 +745,7 @@ func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileSto
 	fh := FileHeader{}
 	err = fh.Load(f)
 	if err != nil {
-		return nil, err
+		return nil, zodb.InvalidTid, err
 	}
 
 	// load index
@@ -778,7 +778,7 @@ func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileSto
 		checkTailGarbage = true
 	}
 	if err != nil {
-		return nil, err
+		return nil, zodb.InvalidTid, err
 	}
 
 	fs.index = index
@@ -789,7 +789,7 @@ func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileSto
 		err = fs.txnhMin.Load(f, txnValidFrom, LoadAll)
 		err = noEOF(err)
 		if err != nil {
-			return nil, err
+			return nil, zodb.InvalidTid, err
 		}
 
 		_ = fs.txnhMax.Load(f, index.TopPos, LoadAll)
@@ -798,19 +798,21 @@ func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileSto
 		// that we read .LenPrev ok.
 		switch fs.txnhMax.LenPrev {
 		case -1:
-			return nil, fmt.Errorf("%s: could not read LenPrev @%d (last transaction)", f.Name(), fs.txnhMax.Pos)
+			return nil, zodb.InvalidTid, fmt.Errorf("%s: could not read LenPrev @%d (last transaction)", f.Name(), fs.txnhMax.Pos)
 		case 0:
-			return nil, fmt.Errorf("%s: could not read LenPrev @%d (last transaction): unexpected EOF backward", f.Name(), fs.txnhMax.Pos)
+			return nil, zodb.InvalidTid, fmt.Errorf("%s: could not read LenPrev @%d (last transaction): unexpected EOF backward", f.Name(), fs.txnhMax.Pos)
 
 		default:
 			// .LenPrev is ok - read last previous record
 			err = fs.txnhMax.LoadPrev(f, LoadAll)
 			if err != nil {
-				return nil, err
+				return nil, zodb.InvalidTid, err
 			}
 		}
 	}
 
+	at0 = fs.txnhMax.Tid
+
 	// there might be simultaneous updates to the data file from outside.
 	// launch the watcher who will observe them.
 	//
@@ -818,12 +820,12 @@ func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileSto
 	// race of missing early file writes.
 	w, err := fsnotify.NewWatcher()
 	if err != nil {
-		return nil, err
+		return nil, zodb.InvalidTid, err
 	}
 	err = w.Add(f.Name())
 	if err != nil {
 		w.Close()	// XXX lclose
-		return nil, err
+		return nil, zodb.InvalidTid, err
 	}
 
 	var errFirstRead chan error
@@ -838,14 +840,14 @@ func Open(ctx context.Context, path string, opt *zodb.DriverOptions) (_ *FileSto
 	if checkTailGarbage {
 		select {
 		case <-ctx.Done():
-			return nil, ctx.Err()
+			return nil, zodb.InvalidTid, ctx.Err()
 
 		case err = <-errFirstRead:
 			if err != nil {
-				return nil, err // it was garbage
+				return nil, zodb.InvalidTid, err // it was garbage
 			}
 		}
 	}
 
-	return fs, nil
+	return fs, at0, nil
 }

go/zodb/storage/fs1/filestorage_test.go

@@ -114,22 +114,22 @@ func checkLoad(t *testing.T, fs *FileStorage, xid zodb.Xid, expect objState) {
 	}
 }
 
-func xfsopen(t testing.TB, path string) *FileStorage {
+func xfsopen(t testing.TB, path string) (*FileStorage, zodb.Tid) {
 	t.Helper()
 	return xfsopenopt(t, path, &zodb.DriverOptions{ReadOnly: true})
 }
 
-func xfsopenopt(t testing.TB, path string, opt *zodb.DriverOptions) *FileStorage {
+func xfsopenopt(t testing.TB, path string, opt *zodb.DriverOptions) (*FileStorage, zodb.Tid) {
 	t.Helper()
-	fs, err := Open(context.Background(), path, opt)
+	fs, at0, err := Open(context.Background(), path, opt)
 	if err != nil {
 		t.Fatal(err)
 	}
-	return fs
+	return fs, at0
 }
 
 func TestLoad(t *testing.T) {
-	fs := xfsopen(t, "testdata/1.fs")
+	fs, _ := xfsopen(t, "testdata/1.fs")
 	defer exc.XRun(fs.Close)
 
 	// current knowledge of what was "before" for an oid as we scan over
@@ -278,7 +278,7 @@ func testIterate(t *testing.T, fs *FileStorage, tidMin, tidMax zodb.Tid, expectv
 }
 
 func TestIterate(t *testing.T) {
-	fs := xfsopen(t, "testdata/1.fs")
+	fs, _ := xfsopen(t, "testdata/1.fs")
 	defer exc.XRun(fs.Close)
 
 	// all []tids in test database
@@ -314,7 +314,7 @@ func TestIterate(t *testing.T) {
 }
 
 func BenchmarkIterate(b *testing.B) {
-	fs := xfsopen(b, "testdata/1.fs")
+	fs, _ := xfsopen(b, "testdata/1.fs")
 	defer exc.XRun(fs.Close)
 
 	ctx := context.Background()
@@ -412,7 +412,10 @@ func TestWatch(t *testing.T) {
 	at := xcommit(0, Object{0, "data0"})
 
 	watchq := make(chan zodb.CommitEvent)
-	fs := xfsopenopt(t, tfs, &zodb.DriverOptions{ReadOnly: true, Watchq: watchq})
+	fs, at0 := xfsopenopt(t, tfs, &zodb.DriverOptions{ReadOnly: true, Watchq: watchq})
+	if at0 != at {
+		t.Fatalf("opened @ %s  ; want %s", at0, at)
+	}
 	ctx := context.Background()
 
 	checkLastTid := func(lastOk zodb.Tid) {
@@ -510,10 +513,13 @@ func TestOpenRecovery(t *testing.T) {
 	}
 	for _, l := range lok {
 		checkL(t, l, func(t *testing.T, tfs string) {
-			fs := xfsopen(t, tfs)
+			fs, at0 := xfsopen(t, tfs)
 			defer func() {
 				err = fs.Close(); X(err)
 			}()
+			if at0 != lastTidOk {
+				t.Fatalf("at0: %s  ; expected: %s", at0, lastTidOk)
+			}
 			head, err := fs.LastTid(ctx); X(err)
 			if head != lastTidOk {
 				t.Fatalf("last_tid: %s  ; expected: %s", head, lastTidOk)
@@ -525,7 +531,7 @@ func TestOpenRecovery(t *testing.T) {
 	// XXX better check 0..sizeof(txnh)-1 but in this range each check is slow.
 	for _, l := range []int{TxnHeaderFixSize-1,1} {
 		checkL(t, l, func(t *testing.T, tfs string) {
-			_, err := Open(ctx, tfs, &zodb.DriverOptions{ReadOnly: true})
+			_, _, err := Open(ctx, tfs, &zodb.DriverOptions{ReadOnly: true})
 			estr := ""
 			if err != nil {
 				estr = err.Error()

go/zodb/storage/fs1/url.go

@@ -27,13 +27,12 @@ import (
 	"lab.nexedi.com/kirr/neo/go/zodb"
 )
 
-func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (zodb.IStorageDriver, error) {
+func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (zodb.IStorageDriver, zodb.Tid, error) {
 	// TODO handle query
 	// XXX u.Path is not always raw path - recheck and fix
 	path := u.Host + u.Path
 
-	fs, err := Open(ctx, path, opt)
-	return fs, err
+	return Open(ctx, path, opt)
 }
 
 func init() {

go/zodb/storage/zeo/zeo.go

@@ -283,7 +283,7 @@ func (r rpc) ereplyf(format string, argv ...interface{}) *errorUnexpectedReply {
 
 // ---- open ----
 
-func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb.IStorageDriver, err error) {
+func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb.IStorageDriver, at0 zodb.Tid, err error) {
 	url := u.String()
 	defer xerr.Contextf(&err, "open %s:", url)
 
@@ -307,11 +307,12 @@ func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb
 	}
 
 	if !opt.ReadOnly {
-		return nil, fmt.Errorf("TODO write mode not implemented")
+		return nil, zodb.InvalidTid, fmt.Errorf("TODO write mode not implemented")
 	}
 
 	// FIXME handle opt.Watchq
 	// for now we pretend as if the database is not changing.
+	// TODO watcher(when implementing): filter-out first < at0 messages.
 	if opt.Watchq != nil {
 		log.Print("zeo: FIXME: watchq support not implemented - there" +
 			  "won't be notifications about database changes")
@@ -319,7 +320,7 @@ func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb
 
 	zl, err := dialZLink(ctx, net, addr)
 	if err != nil {
-		return nil, err
+		return nil, zodb.InvalidTid, err
 	}
 
 	defer func() {
@@ -334,7 +335,7 @@ func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb
 	rpc := z.rpc("register")
 	xlastTid, err := rpc.call(ctx, storageID, opt.ReadOnly)
 	if err != nil {
-		return nil, err
+		return nil, zodb.InvalidTid, err
 	}
 
 	// register returns last_tid in ZEO5 but nothing earlier.
@@ -343,17 +344,24 @@ func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb
 		rpc = z.rpc("lastTransaction")
 		xlastTid, err = rpc.call(ctx)
 		if err != nil {
-			return nil, err
+			return nil, zodb.InvalidTid, err
 		}
 	}
 
 	lastTid, ok := tidUnpack(xlastTid) // XXX -> xlastTid -> scan
 	if !ok {
-		return nil, rpc.ereplyf("got %v; expect tid", xlastTid)
+		return nil, zodb.InvalidTid, rpc.ereplyf("got %v; expect tid", xlastTid)
 	}
 
 	z.lastTid = lastTid
 
+	// XXX since we read lastTid, at least with ZEO < 5, in separate RPC
+	// call, there is a chance, that by the time when lastTid was read some
+	// new transactions were committed. This way lastTid will be > than
+	// some first transactions received by watcher via
+	// "invalidateTransaction" server notification.
+	at0 = lastTid
+
 
 	//call('get_info') -> {}str->str, ex	// XXX can be omitted
 /*
@@ -371,7 +379,7 @@ func openByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (_ zodb
   'supports_record_iternext': True})
 */
 
-	return z, nil
+	return z, at0, nil
 }
 
 func (z *zeo) Close() error {