.

go/neo/client.go

@@ -69,8 +69,8 @@ type Client struct {
 	operational bool // XXX <- somehow move to NodeApp?
 	opReady	    chan struct{} // reinitialized each time state becomes non-operational
 
-	// driver client <- watcher: database commits.
-	watchq chan<- zodb.CommitEvent // FIXME stub
+	// driver client <- watcher: database commits | errors.
+	watchq chan<- zodb.Event // FIXME stub
 }
 
 var _ zodb.IStorageDriver = (*Client)(nil)

go/zodb/db.go

@@ -122,7 +122,7 @@ func NewDB(stor IStorage) *DB {
 		tδkeep: 10*time.Minute, // see δtail discussion
 	}
 
-	watchq := make(chan CommitEvent)
+	watchq := make(chan Event)
 	at0 := stor.AddWatch(watchq)
 	db.δtail = NewΔTail(at0) // init to (at0, at0]
 
@@ -176,7 +176,7 @@ type δwaiter struct {
 // watcher receives events about new committed transactions and updates δtail.
 //
 // it also notifies δtail waiters.
-func (db *DB) watcher(watchq <-chan CommitEvent) { // XXX err ?
+func (db *DB) watcher(watchq <-chan Event) { // XXX err ?
 	for {
 		event, ok := <-watchq
 		if !ok {
@@ -187,13 +187,26 @@ func (db *DB) watcher(watchq <-chan CommitEvent) { // XXX err ?
 
 		fmt.Printf("db: watcher <- %v\n", event)
 
+		var δ *EventCommit
+		switch event := event.(type) {
+		default:
+			panic(fmt.Sprintf("unexepected event: %T", event))
+
+		case *EventError:
+			fmt.Printf("db: watcher: error: %s\n", event.Err)
+			continue // XXX shutdown instead?
+
+		case *EventCommit:
+			δ = event
+		}
+
 		var readyv []chan struct{} // waiters that become ready
 
 		db.mu.Lock()
 
-		db.δtail.Append(event.Tid, event.Changev)
+		db.δtail.Append(δ.Tid, δ.Changev)
 		for w := range db.δwait {
-			if w.at <= event.Tid {
+			if w.at <= δ.Tid {
 				readyv = append(readyv, w.ready)
 				delete(db.δwait, w)
 			}

go/zodb/open.go

@@ -50,9 +50,7 @@ type DriverOptions struct {
 	//
 	// 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
+	Watchq chan<- Event
 }
 
 // DriverOpener is a function to open a storage driver.
@@ -100,7 +98,7 @@ func OpenStorage(ctx context.Context, zurl string, opt *OpenOptions) (IStorage,
 		return nil, fmt.Errorf("zodb: URL scheme \"%s://\" not supported", u.Scheme)
 	}
 
-	drvWatchq := make(chan CommitEvent)
+	drvWatchq := make(chan Event)
 	drvOpt := &DriverOptions{
 		ReadOnly: opt.ReadOnly,
 		Watchq:   drvWatchq,
@@ -131,7 +129,7 @@ func OpenStorage(ctx context.Context, zurl string, opt *OpenOptions) (IStorage,
 		drvWatchq: drvWatchq,
 		drvHead:   at0,
 		watchReq:  make(chan watchRequest),
-		watchTab:  make(map[chan<- CommitEvent]struct{}),
+		watchTab:  make(map[chan<- Event]struct{}),
 
 	}
 	go stor.watcher()	// XXX stop on close
@@ -150,10 +148,10 @@ type storage struct {
 	l1cache *Cache // can be =nil, if opened with NoCache
 
 	// watcher
-	drvWatchq chan CommitEvent                // watchq passed to driver
-	drvHead   Tid                             // last tid received from drvWatchq
-	watchReq  chan watchRequest               // {Add,Del}Watch requests go here
-	watchTab  map[chan<- CommitEvent]struct{} // registered watchers
+	drvWatchq chan Event                // watchq passed to driver
+	drvHead   Tid                       // last tid received from drvWatchq
+	watchReq  chan watchRequest         // {Add,Del}Watch requests go here
+	watchTab  map[chan<- Event]struct{} // registered watchers
 }
 
 // loading goes through cache - this way prefetching can work
@@ -183,9 +181,9 @@ func (s *storage) Prefetch(ctx context.Context, xid Xid) {
 
 // watchRequest represents request to add/del a watch.
 type watchRequest struct {
-	op     watchOp            // add or del
-	ack    chan Tid           // when request processed: at0 for add, ø for del.
-	watchq chan<- CommitEvent // {Add,Del}Watch argument
+	op     watchOp      // add or del
+	ack    chan Tid     // when request processed: at0 for add, ø for del.
+	watchq chan<- Event // {Add,Del}Watch argument
 }
 
 type watchOp int
@@ -222,9 +220,18 @@ func (s *storage) watcher() {
 				return
 			}
 
-			// XXX verify event.Tid ↑  (else e.g. δtail.Append will panic)
-			//     if !↑ - stop the storage with error.
-			s.drvHead = event.Tid
+			switch event := event.(type) {
+			default:
+				panic(fmt.Sprintf("unexpected event: %T", event))
+
+			case *EventError:
+				// ok
+
+			case *EventCommit:
+				// XXX verify event.Tid ↑  (else e.g. δtail.Append will panic)
+				//     if !↑ - stop the storage with error.
+				s.drvHead = event.Tid
+			}
 
 			// deliver event to all watchers
 			for watchq := range s.watchTab {
@@ -236,7 +243,7 @@ func (s *storage) watcher() {
 }
 
 // AddWatch implements Watcher.
-func (s *storage) AddWatch(watchq chan<- CommitEvent) (at0 Tid) {
+func (s *storage) AddWatch(watchq chan<- Event) (at0 Tid) {
 	// XXX when already Closed?
 	ack := make(chan Tid)
 	s.watchReq <- watchRequest{addWatch, ack, watchq}
@@ -244,7 +251,7 @@ func (s *storage) AddWatch(watchq chan<- CommitEvent) (at0 Tid) {
 }
 
 // DelWatch implements Watcher.
-func (s *storage) DelWatch(watchq chan<- CommitEvent) {
+func (s *storage) DelWatch(watchq chan<- Event) {
 	// XXX when already Closed?
 	ack := make(chan Tid)
 	s.watchReq <- watchRequest{delWatch, ack, watchq}

go/zodb/storage/fs1/filestorage.go

@@ -98,8 +98,8 @@ type FileStorage struct {
 	txnhMax TxnHeader // (both with .Len=0 & .Tid=0 if database is empty)
 	downErr error     // !nil when the storage is no longer operational
 
-	// driver client <- watcher: database commits.
-	watchq chan<- zodb.CommitEvent
+	// driver client <- watcher: database commits | errors.
+	watchq chan<- zodb.Event
 
 	down     chan struct{}  // ready when storage is no longer operational
 	downOnce sync.Once      // shutdown may be due to both Close and IO error in watcher
@@ -496,6 +496,9 @@ func (fs *FileStorage) watcher(w *fsnotify.Watcher, errFirstRead chan<- error) {
 	fs.shutdown(err)
 
 	if fs.watchq != nil {
+		if err != nil {
+			fs.watchq <- &zodb.EventError{err}
+		}
 		close(fs.watchq)
 	}
 }
@@ -692,7 +695,7 @@ mainloop:
 				case <-fs.down:
 					return nil
 
-				case fs.watchq <- zodb.CommitEvent{it.Txnh.Tid, δoid}:
+				case fs.watchq <- &zodb.EventCommit{it.Txnh.Tid, δoid}:
 					// ok
 				}
 			}

go/zodb/storage/fs1/filestorage_test.go

@@ -377,7 +377,7 @@ func TestWatch(t *testing.T) {
 	// force tfs creation & open tfs at go side
 	at := xcommit(0, xtesting.ZRawObject{0, b("data0")})
 
-	watchq := make(chan zodb.CommitEvent)
+	watchq := make(chan zodb.Event)
 	fs, at0 := xfsopenopt(t, tfs, &zodb.DriverOptions{ReadOnly: true, Watchq: watchq})
 	if at0 != at {
 		t.Fatalf("opened @ %s  ; want %s", at0, at)
@@ -423,10 +423,22 @@ func TestWatch(t *testing.T) {
 			xtesting.ZRawObject{i, b(datai)})
 
 		// TODO also test for watcher errors
-		e := <-watchq
+		event := <-watchq
 
-		if objvWant := []zodb.Oid{0, i}; !(e.Tid == at && reflect.DeepEqual(e.Changev, objvWant)) {
-			t.Fatalf("watch:\nhave: %s %s\nwant: %s %s", e.Tid, e.Changev, at, objvWant)
+		var δ *zodb.EventCommit
+		switch event := event.(type) {
+		default:
+			panic(fmt.Sprintf("unexpected event: %T", event))
+
+		case *zodb.EventError:
+			t.Fatal(event.Err)
+
+		case *zodb.EventCommit:
+			δ = event
+		}
+
+		if objvWant := []zodb.Oid{0, i}; !(δ.Tid == at && reflect.DeepEqual(δ.Changev, objvWant)) {
+			t.Fatalf("watch:\nhave: %s %s\nwant: %s %s", δ.Tid, δ.Changev, at, objvWant)
 		}
 
 		checkLastTid(at)

go/zodb/storage/zeo/zeo.go

@@ -45,8 +45,8 @@ type zeo struct {
 	mu      sync.Mutex
 	lastTid zodb.Tid
 
-	// driver client <- watcher: database commits.
-	watchq chan<- zodb.CommitEvent // FIXME stub
+	// driver client <- watcher: database commits | errors.
+	watchq chan<- zodb.Event // FIXME stub
 
 	url string // we were opened via this
 }

go/zodb/zodb.go

@@ -428,9 +428,26 @@ type Committer interface {
 	// TpcAbort(txn)
 }
 
+// Event represents one database event.
+//
+// Possible events are:
+//
+//	- EventError	an error happened
+//	- EventCommit	a transaction was committed
+type Event interface {
+	event()
+}
+
+func (_ *EventError)  event() {}
+func (_ *EventCommit) event() {}
 
-// CommitEvent is event describing one observed database commit.
-type CommitEvent struct {
+// EventError is event descrbing an error observed by watcher.
+type EventError struct {
+	Err error
+}
+
+// EventCommit is event describing one observed database commit.
+type EventCommit struct {
 	Tid     Tid   // ID of committed transaction
 	Changev []Oid // ID of objects changed by committed transaction
 }
@@ -453,10 +470,10 @@ type Watcher interface {
 	// Once registered, watchq must be read until DelWatch call.
 	// Not doing so will stuck whole storage.
 	//
-	// Multiple AddWatch calls with the same watchq register watchq only once.	XXX
+	// Registered watchq are closed when the database storage is closed.
 	//
-	// XXX watchq closed when stor.watchq closed?
-	AddWatch(watchq chan<- CommitEvent) (at0 Tid)
+	// Multiple AddWatch calls with the same watchq register watchq only once.	XXX
+	AddWatch(watchq chan<- Event) (at0 Tid)
 
 	// DelWatch unregisters watchq from being notified of database changes.
 	//
@@ -480,7 +497,7 @@ type Watcher interface {
 	// DelWatch is noop if watchq was not registered.
 	//
 	// XXX also return curent head?
-	DelWatch(watchq chan<- CommitEvent)
+	DelWatch(watchq chan<- Event)
 }
 
 

go/zodb/zodbtools/watch.go

@@ -17,9 +17,9 @@
 // See COPYING file for full licensing terms.
 // See https://www.nexedi.com/licensing for rationale and options.
 
-// Zodbwatch - watch database for changes
+// Zodbwatch - watch ZODB database for changes
 //
-// Zodbwatch watches deatbase for changes and prints information about
+// Zodbwatch watches database for changes and prints information about
 // committed transactions. Output formats:
 //
 // Plain:
@@ -67,7 +67,7 @@ func Watch(ctx context.Context, stor zodb.IStorage, w io.Writer, verbose bool) (
 		return err
 	}
 
-	watchq := make(chan zodb.CommitEvent)
+	watchq := make(chan zodb.Event)
 	at0 := stor.AddWatch(watchq)
 	defer stor.DelWatch(watchq)
 
@@ -81,13 +81,24 @@ func Watch(ctx context.Context, stor zodb.IStorage, w io.Writer, verbose bool) (
 		case <-ctx.Done():
 			return ctx.Err()
 
-		case δ, ok := <-watchq:
+		case event, ok := <-watchq:
 			if !ok {
-				// XXX correct?
 				err = emitf("# storage closed")
 				return err
 			}
 
+			var δ *zodb.EventCommit
+			switch event := event.(type) {
+			default:
+				panic(fmt.Sprintf("unexpected event: %T", event))
+
+			case *zodb.EventError:
+				return event.Err
+
+			case *zodb.EventCommit:
+				δ = event
+			}
+
 			err = emitf("txn %s\n", δ.Tid)
 			if err != nil {
 				return err