Unified · Split
| @@ -97,18 +97,20 @@ type Options struct { · 96 unmodified lines | ||
| 97 | 97 | // Perms represents the datafiles modes and permission bits |
| 98 | 98 | DirPerms os.FileMode |
| 99 | 99 | FilePerms os.FileMode |
| 100 | + // RecoverCorruptedTail will attempt to recover a corrupted tail in the last segment automatically. | |
| 101 | + RecoverCorruptedTail bool | |
| 100 | 102 | } |
| 101 | 103 | |
| 102 | 104 | // DefaultOptions for Open(). |
| 103 | 105 | var DefaultOptions = &Options{ |
| 104 | − NoSync: false, // Fsync after every write | |
| 105 | − SegmentSize: 20971520, // 20 MB log segment files | |
| 106 | − LogFormat: Binary, // Binary format is small and fast | |
| 107 | − SegmentCacheSize: 2, // Number of cached in-memory segments | |
| 108 | − NoCopy: false, // Make a new copy of data for every Read call | |
| 109 | − AllowEmpty: false, // Do not allow empty log. 1+ entries required | |
| 110 | − DirPerms: 0750, // Permissions for the created directories | |
| 111 | − FilePerms: 0640, // Permissions for the created data files | |
| 106 | + NoSync: false, // Fsync after every write | |
| 107 | + SegmentSize: 20971520, // 20 MB log segment files. | |
| 108 | + LogFormat: Binary, // Binary format is small and fast. | |
| 109 | + SegmentCacheSize: 2, // Number of cached in-memory segments | |
| 110 | + NoCopy: false, // Make a new copy of data for every Read call. | |
| 111 | + DirPerms: 0750, // Permissions for the created directories | |
| 112 | + FilePerms: 0640, // Permissions for the created data files | |
| 113 | + RecoverCorruptedTail: false, // Don't recover corrupted tail. | |
| 112 | 114 | } |
| 113 | 115 | |
| 114 | 116 | // Log represents a write ahead log |
| @@ -282,15 +284,15 @@ func (l *Log) load() error { · 167 unmodified lines | ||
| 282 | 284 | l.firstIndex = l.segments[0].index |
| 283 | 285 | // Open the last segment for appending |
| 284 | 286 | lseg := l.segments[len(l.segments)-1] |
| 285 | − l.sfile, err = os.OpenFile(lseg.path, os.O_WRONLY, l.opts.FilePerms) | |
| 286 | − if err != nil { | |
| 287 | + // Load the last segment entries | |
| 288 | + if err := l.loadSegmentEntries(lseg, l.opts.RecoverCorruptedTail); err != nil { | |
| 287 | 289 | return err |
| 288 | 290 | } |
| 289 | − if _, err := l.sfile.Seek(0, 2); err != nil { | |
| 291 | + l.sfile, err = os.OpenFile(lseg.path, os.O_WRONLY, l.opts.FilePerms) | |
| 292 | + if err != nil { | |
| 290 | 293 | return err |
| 291 | 294 | } |
| 292 | − // Load the last segment entries | |
| 293 | − if err := l.loadSegmentEntries(lseg); err != nil { | |
| 295 | + if _, err := l.sfile.Seek(int64(len(lseg.ebuf)), 0); err != nil { | |
| 294 | 296 | return err |
| 295 | 297 | } |
| 296 | 298 | l.lastIndex = lseg.index + uint64(len(lseg.epos)) - 1 |
| @@ -555,7 +557,7 @@ func (l *Log) findSegment(index uint64) int { · 258 unmodified lines | ||
| 555 | 557 | return i - 1 |
| 556 | 558 | } |
| 557 | 559 | |
| 558 | −func (l *Log) loadSegmentEntries(s *segment) error { | |
| 560 | +func (l *Log) loadSegmentEntries(s *segment, ignoreCorruptedTail bool) error { | |
| 559 | 561 | data, err := ioutil.ReadFile(s.path) |
| 560 | 562 | if err != nil { |
| 561 | 563 | return err |
| @@ -570,6 +572,9 @@ func (l *Log) loadSegmentEntries(s *segment) error { · 8 unmodified lines | ||
| 570 | 572 | } else { |
| 571 | 573 | n, err = loadNextBinaryEntry(data) |
| 572 | 574 | } |
| 575 | + if err == ErrCorrupt && ignoreCorruptedTail { | |
| 576 | + break | |
| 577 | + } | |
| 573 | 578 | if err != nil { |
| 574 | 579 | return err |
| 575 | 580 | } |
| @@ -577,7 +582,7 @@ func (l *Log) loadSegmentEntries(s *segment) error { · 1 unmodified line | ||
| 577 | 582 | epos = append(epos, bpos{pos, pos + n}) |
| 578 | 583 | pos += n |
| 579 | 584 | } |
| 580 | − s.ebuf = ebuf | |
| 585 | + s.ebuf = ebuf[:pos] | |
| 581 | 586 | s.epos = epos |
| 582 | 587 | return nil |
| 583 | 588 | } |
| @@ -633,7 +638,7 @@ func (l *Log) loadSegment(index uint64) (*segment, error) { · 49 unmodified lines | ||
| 633 | 638 | s := l.segments[idx] |
| 634 | 639 | if len(s.epos) == 0 { |
| 635 | 640 | // load the entries from cache |
| 636 | − if err := l.loadSegmentEntries(s); err != nil { | |
| 641 | + if err := l.loadSegmentEntries(s, false); err != nil { | |
| 637 | 642 | return nil, err |
| 638 | 643 | } |
| 639 | 644 | } |
| @@ -850,7 +855,7 @@ func (l *Log) truncateFront(index uint64) (err error) { · 210 unmodified lines | ||
| 850 | 855 | return err |
| 851 | 856 | } |
| 852 | 857 | // Load the last segment entries |
| 853 | − if err = l.loadSegmentEntries(s); err != nil { | |
| 858 | + if err = l.loadSegmentEntries(s, false); err != nil { | |
| 854 | 859 | return err |
| 855 | 860 | } |
| 856 | 861 | } |
| @@ -958,7 +963,7 @@ func (l *Log) truncateBack(index uint64) (err error) { · 101 unmodified lines | ||
| 958 | 963 | l.segments = append([]*segment{}, l.segments[:segIdx+1]...) |
| 959 | 964 | l.lastIndex = index |
| 960 | 965 | l.clearCache() |
| 961 | − if err = l.loadSegmentEntries(s); err != nil { | |
| 966 | + if err = l.loadSegmentEntries(s, false); err != nil { | |
| 962 | 967 | return err |
| 963 | 968 | } |
| 964 | 969 | return nil |