From 66544660336915373f06d69d261293b737eccc21 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Tue, 24 Sep 2019 12:24:22 -0500
Subject: [PATCH] partially implement truncation of fragments for corrupt ops
 log

Which is to say don't actually implement it, because openStorage
is too messy right now, but this is the rest of the framework,
and now I'm going to digress into fixing openStorage.
---
 fragment.go                 | 21 +++++++++++++++----
 roaring/roaring.go          | 40 +++++++++++++++++++++++++++++++++++++
 roaring/unmarshal_binary.go |  6 +++---
 3 files changed, 60 insertions(+), 7 deletions(-)

diff --git a/fragment.go b/fragment.go
index fa66847b4..85d5cb385 100644
--- a/fragment.go
+++ b/fragment.go
@@ -406,11 +406,24 @@ func (f *fragment) openStorage(unmarshalData bool) error {
 		// *either* or *both* of old and new storage data might be in use.
 		// So we call the thing that should unconditionally unmap both of them...
 		if err := f.storage.UnmarshalBinary(data); err != nil {
-			_, e2 := f.storage.RemapRoaringStorage(nil)
-			if e2 != nil {
-				return fmt.Errorf("unmarshal storage: file=%s, err=%s, clearing old mapping also failed: %v", f.file.Name(), err, e2)
+			// roaring can report advisory-only errors...
+			_, ok := err.(roaring.AdvisoryError)
+			if !ok {
+				name := f.file.Name()
+				f.file.Close()
+				f.file = nil
+				_, e2 := f.storage.RemapRoaringStorage(nil)
+				if e2 != nil {
+					return fmt.Errorf("unmarshal storage: file=%s, err=%s, clearing old mapping also failed: %v", name, err, e2)
+				}
+				return fmt.Errorf("unmarshal storage: file=%s, err=%s", name, err)
+			} else {
+				f.Logger.Printf("warning: unmarshal storage, file=%s, err=%v", f.file.Name(), err)
+			}
+			trunc, ok := err.(roaring.FileShouldBeTruncatedError)
+			if ok {
+				f.Logger.Printf("should probably truncate file %s to %d bytes, but can't yet", f.file.Name(), trunc.SuggestedLength())
 			}
-			return fmt.Errorf("unmarshal storage: file=%s, err=%s", f.file.Name(), err)
 		}
 		f.rowCache = &simpleCache{make(map[uint64]*Row)}
 		f.ops, f.opN = f.storage.Ops()
diff --git a/roaring/roaring.go b/roaring/roaring.go
index 86ee64fc5..046e1f859 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -77,6 +77,46 @@ var containerTypeNames = map[byte]string{
 
 var fullContainer = NewContainerRun([]interval16{{start: 0, last: maxContainerVal}}).Freeze()
 
+// AdvisoryError is used for the special case where we probably want to *report*
+// an error reading a file, but don't want to actually count the file as not
+// being read. For instance, a partial ops-log entry is *probably* harmless;
+// we probably crashed while writing (?) and as such didn't report the write
+// as successful. We hope.
+type AdvisoryError interface {
+	error
+	AdvisoryOnly()
+}
+
+type advisoryError struct {
+	e error
+}
+
+func (a advisoryError) Error() string {
+	return a.e.Error()
+}
+
+// This marks the error as safe to ignore.
+func (a advisoryError) AdvisoryOnly() {
+}
+
+type FileShouldBeTruncatedError interface {
+	AdvisoryError
+	SuggestedLength() int64
+}
+
+type fileShouldBeTruncatedError struct {
+	advisoryError
+	offset int64
+}
+
+func (f *fileShouldBeTruncatedError) SuggestedLength() int64 {
+	return f.offset
+}
+
+func newFileShouldBeTruncatedError(err error, offset int64) *fileShouldBeTruncatedError {
+	return &fileShouldBeTruncatedError{advisoryError: advisoryError{e: err}, offset: offset}
+}
+
 type Containers interface {
 	// Get returns nil if the key does not exist.
 	Get(key uint64) *Container
diff --git a/roaring/unmarshal_binary.go b/roaring/unmarshal_binary.go
index 86f30df0b..c4feb5bd1 100644
--- a/roaring/unmarshal_binary.go
+++ b/roaring/unmarshal_binary.go
@@ -205,15 +205,15 @@ func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
 		// Unmarshal the op and apply it.
 		var opr op
 		if err := opr.UnmarshalBinary(buf); err != nil {
-			// FIXME(benbjohnson): return error with position so file can be trimmed.
-			return err
+			return newFileShouldBeTruncatedError(err, int64(opsOffset))
 		}
 		opr.apply(b)
 		// Increase the op count.
 		b.ops++
 		b.opN += opr.count()
+		opsOffset += opr.size()
 		// Move the buffer forward.
-		buf = buf[opr.size():]
+		buf = buf[opsOffset:]
 	}
 
 	return nil