Add a mechanism for rebuilding meta.db based on segments.

options.go

@@ -66,6 +66,12 @@ func WithMetricsCollector(c metrics.Collector) walOpt {
 	}
 }
 
+func WithRegenerateMeta(b bool) walOpt {
+	return func(w *WAL) {
+		w.regenerateMeta = b
+	}
+}
+
 func (w *WAL) applyDefaultsAndValidate() error {
 	// Check if an external codec has been used that it's not using a reserved ID.
 	if w.codec != nil && w.codec.ID() < FirstExternalCodecID {

segment/filer.go

@@ -104,6 +104,40 @@ func (f *Filer) Open(info types.SegmentInfo) (types.SegmentReader, error) {
 	return openReader(info, rf, &f.bufPool)
 }
 
+// OpenRaw opens an already sealed segment for reading.
+func (f *Filer) OpenRaw(fname string) (*types.SegmentInfo, types.SegmentReader, error) {
+	rf, err := f.vfs.OpenReader(f.dir, fname)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Validate header here since openReader is re-used by writer where it's valid
+	// for the file header not to be committed yet after a crash so we can't check
+	// it there.
+	var hdr [fileHeaderLen]byte
+
+	if _, err := rf.ReadAt(hdr[:], 0); err != nil {
+		if errors.Is(err, io.EOF) {
+			// Treat failure to read a header as corruption since a sealed file should
+			// never not have a valid header. (I.e. even if crashes happen it should
+			// be impossible to seal a segment with no header written so this
+			// indicates that something truncated the file after the fact)
+			return nil, nil, fmt.Errorf("%w: failed to read header: %s", types.ErrCorrupt, err)
+		}
+		return nil, nil, err
+	}
+
+	info, err := readFileHeader(hdr[:])
+	if err != nil {
+		return nil, nil, err
+	}
+	reader, err := openReader(*info, rf, &f.bufPool)
+	if err != nil {
+		return nil, nil, err
+	}
+	return info, reader, nil
+}
+
 // List returns the set of segment IDs currently stored. It's used by the WAL
 // on recovery to find any segment files that need to be deleted following a
 // unclean shutdown. The returned map is a map of ID -> BaseIndex. BaseIndex
@@ -282,3 +316,22 @@ func (f *Filer) DumpLogs(after, before uint64, fn func(info types.SegmentInfo, e
 
 	return nil
 }
+
+func (f *Filer) GetLogs() ([]types.SegmentInfoReader, error) {
+	baseIndexes, segIDsSorted, err := f.listInternal()
+	if err != nil {
+		return nil, err
+	}
+
+	var ret []types.SegmentInfoReader
+	for _, id := range segIDsSorted {
+		fname := fmt.Sprintf(types.SegmentFileNamePattern, baseIndexes[id], id)
+		info, reader, err := f.OpenRaw(fname)
+		if err != nil {
+			return nil, err
+		}
+		ret = append(ret, types.SegmentInfoReader{*info, reader})
+	}
+
+	return ret, nil
+}

types/segment.go

@@ -69,6 +69,11 @@ type SegmentInfo struct {
 	DeletableTime time.Time
 }
 
+type SegmentInfoReader struct {
+	SegmentInfo
+	SegmentReader
+}
+
 // SegmentFiler is the interface that provides access to segments to the WAL. It
 // encapsulated creating, and recovering segments and returning reader or writer
 // interfaces to interact with them. It's main purpose is to abstract the core
@@ -106,6 +111,8 @@ type SegmentFiler interface {
 	// This interface allows a  simpler implementation where we can just delete
 	// the file if it exists without having to scan the underlying storage for a.
 	Delete(baseIndex, ID uint64) error
+
+	GetLogs() ([]SegmentInfoReader, error)
 }
 
 // SegmentWriter manages appending logs to the tail segment of the WAL. It's an

wal.go

@@ -84,6 +84,7 @@ type WAL struct {
 	triggerRotate    chan uint64
 	awaitRotate      chan struct{}
 	requireDeletable bool
+	regenerateMeta   bool
 }
 
 type walOpt func(*WAL)
@@ -112,6 +113,18 @@ func Open(dir string, opts ...walOpt) (*WAL, error) {
 		return nil, err
 	}
 
+	if w.regenerateMeta {
+		w.log.Info("rebuilding meta-db based on segment files")
+		if err := w.rebuildMetaDB(); err != nil {
+			return nil, err
+		}
+
+		persisted, err = w.metaDB.Load(w.dir)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	newState := state{
 		segments:      &immutable.SortedMap[uint64, segmentState]{},
 		nextSegmentID: persisted.NextSegmentID,
@@ -238,6 +251,33 @@ func Open(dir string, opts ...walOpt) (*WAL, error) {
 	return w, nil
 }
 
+func (w *WAL) rebuildMetaDB() error {
+	newState := state{
+		segments: &immutable.SortedMap[uint64, segmentState]{},
+	}
+
+	irs, err := w.sf.GetLogs()
+	if err != nil {
+		return err
+	}
+
+	for _, ir := range irs {
+		newState.segments = newState.segments.Set(ir.SegmentInfo.MinIndex, segmentState{
+			SegmentInfo: ir.SegmentInfo,
+			r:           ir.SegmentReader,
+		})
+		newState.nextSegmentID = ir.SegmentInfo.ID + 1
+	}
+
+	if err := w.metaDB.CommitState(newState.Persistent()); err != nil {
+		return err
+	}
+
+	w.s.Store(&newState)
+
+	return nil
+}
+
 // stateTxn represents a transaction body that mutates the state under the
 // writeLock. s is already a shallow copy of the current state that may be
 // mutated as needed. If a nil error is returned, s will be atomically set as