go/zodb: Fix PActivate not to panic after an error

Persistent.PActivate used to panic when called the second time, if the
first time it hit an error. WCFS hit this in practice via object, that was
previously accessed and pinned into the cache, but later deleted in the
storage.

-> Fix PActivate to reset .loading on an error, so that next time PActivate is
called, it tries to trigger load again instead of panicking. Change doload
criteria from

	state==GHOST && refcnt==1

to

	state==GHOST && loading==nil

because now, after failed PActivate, refcnt can be != 0, if there are several
other PActivate calls that were waiting for the failed PActivate but did not
yet woke up.

Here is how added test fails without the fix:

    --- FAIL: TestActivateAfterDelete (1.65s)
    panic: t.zodb.MyObject(0000000000000065): activate: need to load, but .loading != nil [recovered]
            panic: t.zodb.MyObject(0000000000000065): activate: need to load, but .loading != nil

    goroutine 10085 [running]:
    testing.tRunner.func1.2(0x649020, 0xc000520660)
            /home/kirr/src/tools/go/go/src/testing/testing.go:1143 +0x332
    testing.tRunner.func1(0xc0001cb080)
            /home/kirr/src/tools/go/go/src/testing/testing.go:1146 +0x4b6
    panic(0x649020, 0xc000520660)
            /home/kirr/src/tools/go/go/src/runtime/panic.go:965 +0x1b9
    lab.nexedi.com/kirr/neo/go/zodb.(*Persistent).PActivate(0xc0001184d0, 0x6e8360, 0xc00019ac90, 0x0, 0x0)
            /home/kirr/src/neo/src/lab.nexedi.com/kirr/neo/go/zodb/persistent.go:191 +0xce5
    lab.nexedi.com/kirr/neo/go/zodb.TestActivateAfterDelete(0xc0001cb080)
            /home/kirr/src/neo/src/lab.nexedi.com/kirr/neo/go/zodb/persistent_test.go:786 +0x72c

go/internal/xtesting/xtesting.go

@@ -95,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)
@@ -112,10 +113,15 @@ 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 {
+		// !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")
 
 	// run py `zodb commit`

go/zodb/impexp_test.go

-// Copyright (C) 2019  Nexedi SA and Contributors.
+// 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
@@ -26,9 +26,15 @@ import (
 // imported at runtime via import_x_test due to cyclic dependency:
 
 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.
+// 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
@@ -34,6 +34,7 @@ import (
 
 func init() {
 	zodb.ZPyCommit    = ZPyCommit
+	zodb.ZPyCommitRaw = ZPyCommitRaw
 }
 
 // ZPyCommit commits new transaction with specified objects.
@@ -41,7 +42,7 @@ 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 rawobjv []zodb.ZRawObject // raw zodb objects data to commit
 	var bufv []*mem.Buf           // buffers to release
 	defer func() {
 		for _, buf := range bufv {
@@ -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()

go/zodb/persistent_test.go

@@ -21,6 +21,7 @@ package zodb
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -329,6 +330,18 @@ func (t *tDB) Commit() Tid {
 	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.
 func (t *tDB) Open(opt *ConnOptions) *tConnection {
 	t.Helper()
@@ -719,6 +732,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)