Repository: etcd-io/etcd
File: cache/cache_test.go (Go)

Code before:
	return nil, nil
}

func (kvStub) Delete(ctx context.Context, key string, _ ...clientv3.OpOption) (*clientv3.DeleteResponse, error) {
	return nil, nil
}

func (kvStub) Compact(ctx context.Context, rev int64, _ ...clientv3.CompactOption) (*clientv3.CompactResponse, error) {
	return nil, nil
}

func (kvStub) Do(ctx context.Context, op clientv3.Op) (clientv3.OpResponse, error) {
	return clientv3.OpResponse{}, nil
}

func (kvStub) Txn(ctx context.Context) clientv3.Txn {
	return nil
}

func newMockWatcher(buf int) *mockWatcher {
	return &mockWatcher{
		responses:  make(chan clientv3.WatchResponse, buf),
		registered: make(chan struct{}),
	}
}

func (m *mockWatcher) Watch(_ context.Context, _ string, _ ...clientv3.OpOption) clientv3.WatchChan {
	select { // prevent “close of closed channel” on re-watch
	case <-m.registered:
	default:
		close(m.registered)
	}
	return m.responses
}

func (m *mockWatcher) RequestProgress(_ context.Context) error { return nil }

func (m *mockWatcher) Close() error { close(m.responses); return nil }

func event(eventType mvccpb.Event_EventType, key string, rev int64) *clientv3.Event {
	return &clientv3.Event{
		Type: eventType,
		Kv: &mvccpb.KeyValue{
			Key:            []byte(key),
			ModRevision:    rev,
			CreateRevision: rev,
			Version:        1,
		},
	}
}


Review comment (bug):
Note for the future. If the underlying watch is broken/compacted, we could still serve serializable requests for data we have cached. Dropping is just needed before we start applying new data.

Code after:
		Watcher: newMockWatcher(16),
		KV:      newKVStub(firstSnapshot, secondSnapshot),
	}
	cache, err := New(fakeClient, "")
	if err != nil {
		t.Fatalf("New cache: %v", err)
	}
	defer cache.Close()
	mw := fakeClient.Watcher.(*mockWatcher)

	t.Log("Phase 1: initial getWatch bootstrap")
	mw.triggerCreatedNotify()
	<-mw.registered
	if err = cache.WaitReady(t.Context()); err != nil {
		t.Fatalf("initial WaitReady: %v", err)
	}

	t.Log("Phase 2: simulate compaction")
	mw.errorCompacted(10)

	waitUntil(t, time.Second, 10*time.Millisecond, func() bool { return !cache.Ready() })
	start := time.Now()

	ctxGet, cancelGet := context.WithTimeout(t.Context(), 100*time.Millisecond)
	defer cancelGet()
	_, err = cache.Get(ctxGet, "foo", clientv3.WithSerializable())
	if !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("expected Get() to timeout waiting for ready after compaction, got %v", err)
	}

	t.Log("Phase 3: resync after compaction")
	mw.triggerCreatedNotify()
	if err = cache.WaitReady(t.Context()); err != nil {
		t.Fatalf("second WaitReady: %v", err)
	}
	elapsed := time.Since(start)
	if elapsed > time.Second {
		t.Fatalf("cache was unready for %v; want:  < 1 s", elapsed)
	}

	expectSnapshotRev := int64(20)
	expectedWatchStart := secondSnapshot.Header.Revision + 1
	if gotWatchStart := mw.lastStartRev; gotWatchStart != expectedWatchStart {
		t.Errorf("Watch started at rev=%d; want %d", gotWatchStart, expectedWatchStart)
	}

	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	if err = cache.WaitForRevision(ctx, expectSnapshotRev); err != nil {
		t.Fatalf("cache never reached rev=%d: %v", expectSnapshotRev, err)
	}