manifest: added TestManifestInitCorruptedBlock

manifest/manifest_manager.go

@@ -220,17 +220,10 @@ func (m *Manager) flushPendingEntriesLocked(ctx context.Context) (string, error)
 	}
 
 	var buf bytes.Buffer
-
 	gz := gzip.NewWriter(&buf)
-	if err := json.NewEncoder(gz).Encode(man); err != nil {
-		return "", fmt.Errorf("unable to marshal: %v", err)
-	}
-	if err := gz.Flush(); err != nil {
-		return "", fmt.Errorf("unable to flush: %v", err)
-	}
-	if err := gz.Close(); err != nil {
-		return "", fmt.Errorf("unable to close: %v", err)
-	}
+	mustSucceed(json.NewEncoder(gz).Encode(man))
+	mustSucceed(gz.Flush())
+	mustSucceed(gz.Close())
 
 	blockID, err := m.b.WriteBlock(ctx, buf.Bytes(), manifestBlockPrefix)
 	if err != nil {
@@ -247,6 +240,12 @@ func (m *Manager) flushPendingEntriesLocked(ctx context.Context) (string, error)
 	return blockID, nil
 }
 
+func mustSucceed(e error) {
+	if e != nil {
+		panic("unexpected failure: " + e.Error())
+	}
+}
+
 // Delete marks the specified manifest ID for deletion.
 func (m *Manager) Delete(ctx context.Context, id string) error {
 	if err := m.ensureInitialized(ctx); err != nil {

manifest/manifest_manager_test.go

@@ -123,6 +123,80 @@ func TestManifest(t *testing.T) {
 	verifyItemNotFound(ctx, t, mgr3, id3)
 }
 
+func TestManifestInitCorruptedBlock(t *testing.T) {
+	ctx := context.Background()
+	data := map[string][]byte{}
+	st := storagetesting.NewMapStorage(data, nil, nil)
+
+	f := block.FormattingOptions{
+		BlockFormat: "UNENCRYPTED_HMAC_SHA256_128",
+		MaxPackSize: 100000,
+	}
+
+	// write some data to storage
+	bm, err := block.NewManager(ctx, st, f, block.CachingOptions{})
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	mgr, err := NewManager(ctx, bm)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	mgr.Put(ctx, map[string]string{"type": "foo"}, map[string]string{"some": "value"})
+	mgr.Flush(ctx)
+	bm.Flush(ctx)
+
+	// corrupt data at the storage level.
+	for k, v := range data {
+		if strings.HasPrefix(k, "p") {
+			for i := 0; i < len(v); i++ {
+				v[i] ^= 1
+			}
+		}
+	}
+
+	// make a new block manager based on corrupted data.
+	bm, err = block.NewManager(ctx, st, f, block.CachingOptions{})
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	mgr, err = NewManager(ctx, bm)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	cases := []struct {
+		desc string
+		f    func() error
+	}{
+		{"GetRaw", func() error { _, err := mgr.GetRaw(ctx, "anything"); return err }},
+		{"GetMetadata", func() error { _, err := mgr.GetMetadata(ctx, "anything"); return err }},
+		{"Get", func() error { return mgr.Get(ctx, "anything", nil) }},
+		{"Delete", func() error { return mgr.Delete(ctx, "anything") }},
+		{"Find", func() error { _, err := mgr.Find(ctx, nil); return err }},
+		{"Put", func() error {
+			_, err := mgr.Put(ctx, map[string]string{
+				"type": "foo",
+			}, map[string]string{
+				"some": "value",
+			})
+			return err
+		}},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.desc, func(t *testing.T) {
+			err := tc.f()
+			if err == nil || !strings.Contains(err.Error(), "invalid checksum") {
+				t.Errorf("invalid error when initializing malformed manifest manager: %v", err)
+			}
+		})
+	}
+}
+
 func addAndVerify(ctx context.Context, t *testing.T, mgr *Manager, labels map[string]string, data map[string]int) string {
 	t.Helper()
 	id, err := mgr.Put(ctx, labels, data)