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

@@ -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()

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

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