integeration: fix TestV3HashKV hash mismatch

[fanminshi]

TestV3HashKV returns hash mismatch because UnsafeForEach traverses mvcc
at a different order depends on if there are uncommitted items or not.
Before uncommitted items are committed, UnsafeForEach traverses those first before items in boltdb.
After uncommitted items have been committed, UnsafeForEach traverses boltdb first then the recently committed items.
Hence, hash on same items with different order results hash mismatch.

[heyitsanthony]

f1 doesn't do anything; can replace with visitor to simplify. Can also drop the extra unlock since lock/unlock should typically be paired when convenient, plus have more descriptive names since more complicated now:

func (rt *readTx) UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error {
	dups := make(map[string]struct{})
	getDups := func(k, v []byte) error {
		dups[string(k)] = struct{}{}
		return nil
	}
	visitNoDup := func(k, v []byte) error {
		if _, ok := dups[string(k)]; ok {
			return nil
		}
		return visitor(k, v)
	}
	if err := rt.buf.ForEach(bucketName, getDups); err != nil {
		return err
	}
	rt.txmu.Lock()
	err := unsafeForEach(rt.tx, bucketName, visitNoDup)
	rt.txmu.Unlock()
	if err != nil {
		return err
	}
	return rt.buf.ForEach(bucketName, visitor)
}

[heyitsanthony]

There should be a backend test that exercises the bug; triggering it at this level is too indirect.

Here's the test I wrote to confirm this patch works as expected:

// TestBackendWritebackForEach checks that partially written / buffered                                                                                      
// data is visited in the same order as fully committed data.                                                                                                
func TestBackendWritebackForEach(t *testing.T) {                                                                                                             
        b, tmpPath := NewTmpBackend(time.Hour, 10000)                                                                                                        
        defer cleanup(b, tmpPath)                                                                                                                            
                                                                                                                                                             
        tx := b.BatchTx()
        tx.Lock()
        tx.UnsafeCreateBucket([]byte("key"))
        for i := 0; i < 5; i++ {
                k := []byte(fmt.Sprintf("%04d", i))                                                                                                          
                tx.UnsafePut([]byte("key"), k, []byte("bar"))                                                                                                
        }
        tx.Unlock()                                                                                                                                          
        
        // writeback                                                                                                                                         
        b.ForceCommit()
        
        tx.Lock()
        tx.UnsafeCreateBucket([]byte("key"))                                                                                                                 
        for i := 5; i < 20; i++ {                                                                                                                            
                k := []byte(fmt.Sprintf("%04d", i))                                                                                                          
                tx.UnsafePut([]byte("key"), k, []byte("bar"))                                                                                                
        }       
        tx.Unlock()
        
        seq := ""                                                                                                                                            
        getSeq := func(k, v []byte) error {
                seq += string(k)
                return nil
        }       
        rtx := b.ReadTx()
        rtx.Lock()      
        rtx.UnsafeForEach([]byte("key"), getSeq)                                                                                                             
        rtx.Unlock() 
                
        partialSeq := seq
                
        seq = ""
        b.ForceCommit()                                                                                                                                      

        tx.Lock()                                                                                                                                            
        tx.UnsafeForEach([]byte("key"), getSeq)                                                                                                              
        tx.Unlock()
        
        if seq != partialSeq {                                                                                                                               
                t.Fatalf("expected %q, got %q", seq, partialSeq)                                                                                             
        }
}

[heyitsanthony]

please keep this integration test as-is; the backend unit test is enough

[fanminshi]

sounds good.

[fanminshi]

all fixed. PTAL.

[heyitsanthony]

lgtm once CI passes

[fanminshi]

via https://jenkins-etcd-public.prod.coreos.systems/job/etcd-coverage/1851/console

# github.com/coreos/etcd
/usr/local/go/pkg/tool/linux_amd64/link: running gcc failed: exit status 1
collect2: error: ld returned 1 exit status

this is failure doesn't seem to be related to this pr.

[heyitsanthony]

it's probably from the CI machine being full

[fanminshi]

both ci failures are unrelated; proxy-ci failure #8330.