Watch channel closed unexpectedly

[nych]

Description

When performing a read query that initially returns no results, the watch channel returned by the query closes on any subsequent write transaction. This happens even if the write transaction does not affect the result of the initial query.

Steps to Reproduce

1. Perform a read query.
2. Obtain the watch channel for the query.
3. Perform any write transaction on the DB.
4. Observe that the watch channel closes.

Expected Behavior

The watch channel should remain open until a write transaction specifically affects the result of the query it was opened for.

Actual Behavior

The watch channel closes on any subsequent write transaction, even if the write does not affect the result of the query for which the channel was opened.

Environment

• OS: Linux 5.15.0-86-generic
• Go Version: go1.17.12

Code Sample

package main

import (
	"fmt"
	"log"
	"sync"

	"github.com/hashicorp/go-memdb"
)

type Person struct {
	Email string
	Name  string
	Age   int
}

// Create the DB schema
func schema() *memdb.DBSchema {
	return &memdb.DBSchema{
		Tables: map[string]*memdb.TableSchema{
			"person": {
				Name: "person",
				Indexes: map[string]*memdb.IndexSchema{
					"id": {
						Name:    "id",
						Unique:  true,
						Indexer: &memdb.StringFieldIndex{Field: "Email"},
					},
				},
			},
		},
	}
}

func main() {
	db, err := memdb.NewMemDB(schema())
	if err != nil {
		panic(err)
	}

	var (
		joesWC   <-chan struct{}
		luciesWC <-chan struct{}
	)
	{
		txn := db.Txn(false)
		if it, err := txn.Get("person", "id", "[email protected]"); err != nil {
			panic(err)
		} else if it.Next() != nil {
			panic(fmt.Errorf("did not expect any results"))
		} else {
			joesWC = it.WatchCh()
		}

		if it, err := txn.Get("person", "id", "[email protected]"); err != nil {
			panic(err)
		} else if it.Next() != nil {
			panic(fmt.Errorf("did not expect any results"))
		} else {
			luciesWC = it.WatchCh()
		}
	}
	if joesWC == nil || luciesWC == nil { // make sure we have no nil channels by any chance
		panic("watch channel not initialized")
	}
	{
		txn := db.Txn(true)
		defer txn.Abort()
		if err := txn.Insert("person", Person{
			Email: "[email protected]",
		}); err != nil {
			panic(err)
		}
		txn.Commit()
	}

	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		log.Println("wait for joes trigger")
		<-joesWC
		log.Println("received joes trigger") // this is expected
	}()
	wg.Add(1)
	go func() {
		defer wg.Done()
		log.Println("wait for lucies trigger")
		<-luciesWC
		log.Println("received lucies trigger") // this is NOT expected
	}()

	wg.Wait()
}