From 30fd873a87379232e325efae2dd743ce00d9c576 Mon Sep 17 00:00:00 2001
From: Jon Seymour <jon@wildducktheories.com>
Date: Tue, 16 Feb 2016 12:54:55 +1100
Subject: [PATCH] tsm: add TestCache_CacheRollback

The test now demonstrates that if a snapshot is rolled back, the failed
snapshots (only) will appear in the slice returned by the next
PrepareSnapshots call.

The test demonstrates how the discovery of new WAL segment leads to the
new files being associated with the latest snapshot.

The test demonstrates that when retrying multiple snapshots, we
can successfully commit partial progress, even if we need to rollback
one of the later snapshots.
---
 tsdb/engine/tsm1/cache_test.go | 88 +++++++++++++++++++++++++++++++++-
 1 file changed, 87 insertions(+), 1 deletion(-)

diff --git a/tsdb/engine/tsm1/cache_test.go b/tsdb/engine/tsm1/cache_test.go
index d8c41187dc1..064cd20813b 100644
--- a/tsdb/engine/tsm1/cache_test.go
+++ b/tsdb/engine/tsm1/cache_test.go
@@ -165,6 +165,92 @@ func TestCache_CacheSnapshot(t *testing.T) {
 	if deduped := c.Values("foo"); !reflect.DeepEqual(expValues, deduped) {
 		t.Fatalf("post-commit values for foo incorrect, exp: %v, got %v", expValues, deduped)
 	}
+}
+
+func TestCache_CacheRollbackSnapshots(t *testing.T) {
+	segment2 := "segment2.wal"
+	segment3 := "segment3.wal"
+	segment4 := "segment4.wal"
+	segment5 := "segment5.wal"
+	v4 := NewValue(time.Unix(6, 0).UTC(), 5.0)
+	v5 := NewValue(time.Unix(1, 0).UTC(), 5.0)
+	v6 := NewValue(time.Unix(7, 0).UTC(), 5.0)
+
+	c := NewCache(512)
+
+	// Write a new value to the cache.
+	if err := c.Write("foo", Values{v4, v5}); err != nil {
+		t.Fatalf("failed to write value, key foo to cache: %s", err.Error())
+	}
+
+	expValues := Values{v5, v4}
+	if deduped := c.Values("foo"); !reflect.DeepEqual(expValues, deduped) {
+		t.Fatalf("pre-snapshot write values for foo incorrect, exp: %v, got %v", expValues, deduped)
+	}
+
+	snapshots := c.PrepareSnapshots([]string{segment2})
+
+	if err := c.Write("foo", Values{v6}); err != nil {
+		t.Fatalf("failed to write post-prepare-snapshots key foo to cache: %s", err.Error())
+	}
+
+	expValues = Values{v5, v4, v6}
+	if deduped := c.Values("foo"); !reflect.DeepEqual(expValues, deduped) {
+		t.Fatalf("post-prepare-snapshots values for foo incorrect, exp: %v, got %v", expValues, deduped)
+	}
+
+	c.RollbackSnapshots(snapshots)
+
+	expValues = Values{v5, v4, v6}
+	if deduped := c.Values("foo"); !reflect.DeepEqual(expValues, deduped) {
+		t.Fatalf("post-rollback values for foo incorrect, exp: %v, got %v", expValues, deduped)
+	}
+
+	snapshots = c.PrepareSnapshots([]string{segment2, segment3, segment4})
+	expLength := 2
+	if length := len(snapshots); length != expLength {
+		t.Fatalf("post-prepare-snap-shots: length of snaphots incorrect, exp: %v, got %v", expLength, length)
+	}
+
+	expFiles := [][]string{[]string{segment2}, []string{segment3, segment4}}
+	if files := [][]string{snapshots[0].Files(), snapshots[1].Files()}; !reflect.DeepEqual(expFiles, files) {
+		t.Fatalf("files is incorrect, exp: %v, got %v", expFiles, files)
+	}
+
+	expFooValues := []Values{Values{v5, v4}, Values{v6}}
+	if fooValues := []Values{snapshots[0].Values("foo"), snapshots[1].Values("foo")}; !reflect.DeepEqual(expFooValues, fooValues) {
+		t.Fatalf("snapshot values are incorrect, exp: %v, got %v", expFooValues, fooValues)
+	}
+
+	snapshots[0] = nil
+	c.RollbackSnapshots(snapshots)
+
+	expValues = Values{v6}
+	if deduped := c.Values("foo"); !reflect.DeepEqual(expValues, deduped) {
+		t.Fatalf("post-rollback values for foo incorrect, exp: %v, got %v", expValues, deduped)
+	}
+
+	snapshots = c.PrepareSnapshots([]string{segment3, segment4, segment5})
+	expLength = 2
+	if length := len(snapshots); length != expLength {
+		t.Fatalf("post-prepare-snap-shots: length of snaphots incorrect, exp: %v, got %v", expLength, length)
+	}
+
+	expFiles = [][]string{[]string{segment3, segment4}, []string{segment5}}
+	if files := [][]string{snapshots[0].Files(), snapshots[1].Files()}; !reflect.DeepEqual(expFiles, files) {
+		t.Fatalf("snapshot files are incorrect, exp: %v, got %v", expFiles, files)
+	}
+
+	expFooValues = []Values{Values{v6}, nil}
+	if fooValues := []Values{snapshots[0].Values("foo"), snapshots[1].Values("foo")}; !reflect.DeepEqual(expFooValues, fooValues) {
+		t.Fatalf("snapshot values are incorrect, exp: %v, got %v", expFooValues, fooValues)
+	}
+
+	c.CommitSnapshots()
+
+	expValues = nil
+	if deduped := c.Values("foo"); !reflect.DeepEqual(expValues, deduped) {
+		t.Fatalf("post-commit values for foo incorrect, exp: %v, got %v", expValues, deduped)
 	}
 }
 
@@ -173,7 +259,7 @@ func TestCache_CacheEmptySnapshot(t *testing.T) {
 
 	// Grab snapshot, and ensure it's as expected.
 	snapshots := c.PrepareSnapshots([]string{"foo.wal"})
-	if deduped := snapshots[0].values("foo"); !reflect.DeepEqual(Values(nil), deduped) {
+	if deduped := snapshots[0].Values("foo"); !reflect.DeepEqual(Values(nil), deduped) {
 		t.Fatalf("snapshotted values for foo incorrect, exp: %v, got %v", nil, deduped)
 	}