func (d *diskQueueWriter) TryFixWriteEnd(offset BackendOffset, totalCnt int64) (diskQueueEndInfo, error) {
	d.Lock()
	defer d.Unlock()
	if offset < d.diskQueueStart.virtualEnd || totalCnt < d.diskQueueStart.TotalMsgCnt() {
		nsqLog.Logf("reset write end to %v:%v invalid, less than queue start %v", offset, totalCnt, d.diskQueueStart)
		return d.diskWriteEnd, ErrInvalidOffset
	}
	if offset < d.diskWriteEnd.Offset() {
		return d.diskWriteEnd, ErrInvalidOffset
	}
	if d.needSync {
		d.syncAll(true)
	}
	nsqLog.Logf("fix write end from %v to %v, reset to totalCnt: %v", d.diskWriteEnd.Offset(), offset, totalCnt)
	newEnd := d.diskQueueStart.Offset()
	newWriteFileNum := d.diskQueueStart.EndOffset.FileNum
	newWritePos := d.diskQueueStart.EndOffset.Pos
	for {
		f, err := os.Stat(d.fileName(newWriteFileNum))
		if err != nil {
			nsqLog.LogErrorf("stat data file error %v, %v", offset, newWriteFileNum)
			return d.diskWriteEnd, err
		}
		diff := offset - newEnd
		if newWritePos+int64(diff) <= f.Size() {
			newWritePos += int64(diff)
			newEnd = offset
			break
		}
		newEnd += BackendOffset(f.Size() - newWritePos)
		newWritePos = 0
		newWriteFileNum++
	}
	d.diskWriteEnd.EndOffset.FileNum = newWriteFileNum
	d.diskWriteEnd.EndOffset.Pos = newWritePos
	d.diskWriteEnd.virtualEnd = offset
	atomic.StoreInt64(&d.diskWriteEnd.totalMsgCnt, int64(totalCnt))
	d.diskReadEnd = d.diskWriteEnd
	d.closeCurrentFile()
	nsqLog.Logf("fix write end result : %v", d.diskWriteEnd)
	d.truncateDiskQueueToWriteEnd()
	return d.diskWriteEnd, nil
}
TryFixWriteEnd should only be used to expand current write end, for truncate use resetwrite end instead
package nsqd
import (
	"bufio"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"math/rand"
	"os"
	"path"
	"sync"
	"sync/atomic"
	"time"
	"github.com/youzan/nsq/internal/util"
)
const (
	MAX_QUEUE_OFFSET_META_DATA_KEEP = 100
)
var (
	ErrInvalidOffset     = errors.New("invalid offset")
	ErrNeedFixQueueStart = errors.New("init queue start should be fixed")
	ErrNeedFixQueueEnd   = errors.New("init queue end should be fixed")
	writeBufSize         = 1024 * 128
)
func getQueueFileOffsetMeta(dataFileName string) (int64, int64, int64, error) {
	fName := dataFileName + ".offsetmeta.dat"
	f, err := os.OpenFile(fName, os.O_RDONLY, 0644)
	if err != nil {
		return 0, 0, 0, err
	}
	defer f.Close()
	cnt := int64(0)
	startPos := int64(0)
	endPos := int64(0)
	_, err = fmt.Fscanf(f, "%d\n%d,%d\n",
		&cnt,
		&startPos, &endPos)
	if err != nil {
		nsqLog.LogErrorf("failed to read offset meta (%v): %v", fName, err)
		return 0, 0, 0, err
	}
	return cnt, startPos, endPos, nil
}
type diskQueueWriter struct {
	diskWriteEnd diskQueueEndInfo
	diskReadEnd  diskQueueEndInfo
	diskQueueStart diskQueueEndInfo
	sync.RWMutex
	name            string
	dataPath        string
	maxBytesPerFile int64  
	minMsgSize      int32
	maxMsgSize      int32
	exitFlag        int32
	needSync        bool
	writeFile     *os.File
	bufferWriter  *bufio.Writer
	bufSize       int64
	metaFlushLock sync.RWMutex
	metaStorage   IMetaStorage
	readOnly      bool
}
type extraMeta struct {
	SegOffset     diskQueueOffset
	VirtualOffset BackendOffset
	TotalMsgCnt   int64
}
func newDiskQueueWriterWithMetaStorage(name string, dataPath string, maxBytesPerFile int64,
	minMsgSize int32, maxMsgSize int32,
	syncEvery int64, metaS IMetaStorage) (BackendQueueWriter, error) {
	return newDiskQueueWriter(name, dataPath, maxBytesPerFile,
		minMsgSize, maxMsgSize, syncEvery, false, metaS)
}
func NewDiskQueueWriterForRead(name string, dataPath string, maxBytesPerFile int64,
	minMsgSize int32, maxMsgSize int32,
	syncEvery int64, metaS IMetaStorage) (BackendQueueWriter, error) {
	return newDiskQueueWriter(name, dataPath, maxBytesPerFile,
		minMsgSize, maxMsgSize, syncEvery, true, metaS)
}
func newDiskQueueWriter(name string, dataPath string, maxBytesPerFile int64,
	minMsgSize int32, maxMsgSize int32,
	syncEvery int64, readOnly bool, metaS IMetaStorage) (BackendQueueWriter, error) {
	d := diskQueueWriter{
		name:            name,
		dataPath:        dataPath,
		maxBytesPerFile: maxBytesPerFile,
		minMsgSize:      minMsgSize,
		maxMsgSize:      maxMsgSize,
		metaStorage:     metaS,
	}
	if d.metaStorage == nil {
		d.metaStorage = &fileMetaStorage{}
	}
	d.readOnly = readOnly
	err := d.retrieveMetaData(readOnly)
	if err != nil && !os.IsNotExist(err) {
		nsqLog.LogErrorf("diskqueue(%s) failed to retrieveMetaData - %s", d.name, err)
		return &d, err
	}
	err = d.initQueueReadStart()
	if err != nil && !os.IsNotExist(err) {
		nsqLog.LogErrorf("diskqueue(%s) failed to init queue start- %s", d.name, err)
		return &d, err
	}
	if !readOnly {
		if d.diskQueueStart.EndOffset.GreatThan(&d.diskWriteEnd.EndOffset) ||
			d.diskQueueStart.Offset() > d.diskWriteEnd.Offset() {
			nsqLog.LogErrorf("diskqueue(%s) queue start invalid: %v, write end: %v",
				d.name, d.diskQueueStart, d.diskWriteEnd)
			_, err := os.Stat(d.fileName(d.diskWriteEnd.EndOffset.FileNum))
			if err != nil {
				if d.diskWriteEnd.EndOffset.FileNum <= 0 || d.diskWriteEnd.EndOffset.Pos > 0 {
					return &d, ErrNeedFixQueueEnd
				}
				_, err := os.Stat(d.fileName(d.diskWriteEnd.EndOffset.FileNum - 1))
				if err != nil {
					return &d, ErrNeedFixQueueEnd
				}
			}
			if d.diskWriteEnd.EndOffset.FileNum == 0 &&
				d.diskWriteEnd.EndOffset.Pos == 0 {
				d.diskQueueStart.EndOffset = d.diskWriteEnd.EndOffset
			} else if d.diskWriteEnd.Offset() == 0 {
				d.diskQueueStart = d.diskWriteEnd
			} else if d.diskWriteEnd.EndOffset.FileNum == 0 {
				d.diskQueueStart = diskQueueEndInfo{}
			} else {
				return &d, ErrNeedFixQueueStart
			}
		}
		d.saveExtraMeta()
	}
	return &d, nil
}
func (d *diskQueueWriter) tryFixData() error {
	d.Lock()
	defer d.Unlock()
	err := d.initQueueReadStart()
	if err == ErrNeedFixQueueStart {
		d.diskQueueStart = diskQueueEndInfo{}
	}
	d.saveExtraMeta()
	if err != nil && !os.IsNotExist(err) {
		return err
	}
	return nil
}
func (d *diskQueueWriter) SetBufSize(s int64) {
	atomic.StoreInt64(&d.bufSize, s)
}
func (d *diskQueueWriter) PutV2(data []byte) (BackendOffset, int32, diskQueueEndInfo, error) {
	d.Lock()
	if d.exitFlag == 1 {
		d.Unlock()
		return 0, 0, diskQueueEndInfo{}, errors.New("exiting")
	}
	offset, writeBytes, dend, werr := d.writeOne(data, false, 0)
	var e diskQueueEndInfo
	if dend != nil {
		e = *dend
	}
	d.Unlock()
	return offset, writeBytes, e, werr
}
func (d *diskQueueWriter) PutRawV2(data []byte, msgCnt int32) (BackendOffset, int32, diskQueueEndInfo, error) {
	d.Lock()
	if d.exitFlag == 1 {
		d.Unlock()
		return 0, 0, diskQueueEndInfo{}, errors.New("exiting")
	}
	offset, writeBytes, dend, werr := d.writeOne(data, true, msgCnt)
	var e diskQueueEndInfo
	if dend != nil {
		e = *dend
	}
	d.Unlock()
	return offset, writeBytes, e, werr
}
func (d *diskQueueWriter) Put(data []byte) (BackendOffset, int32, int64, error) {
	d.Lock()
	if d.exitFlag == 1 {
		d.Unlock()
		return 0, 0, 0, errors.New("exiting")
	}
	offset, writeBytes, dend, werr := d.writeOne(data, false, 0)
	var e diskQueueEndInfo
	if dend != nil {
		e = *dend
	}
	d.Unlock()
	return offset, writeBytes, e.TotalMsgCnt(), werr
}
func (d *diskQueueWriter) RollbackWriteV2(offset BackendOffset, diffCnt uint64) (diskQueueEndInfo, error) {
	d.Lock()
	defer d.Unlock()
	if offset < d.diskQueueStart.virtualEnd ||
		d.diskWriteEnd.TotalMsgCnt()-int64(diffCnt) < d.diskQueueStart.TotalMsgCnt() {
		nsqLog.Logf("rollback write to %v:%v invalid, less than queue start %v", offset, diffCnt, d.diskQueueStart)
		return d.diskWriteEnd, ErrInvalidOffset
	}
	if d.needSync {
		d.syncAll(true)
	}
	if offset > d.diskWriteEnd.Offset() {
		return d.diskWriteEnd, ErrInvalidOffset
	}
	if offset < d.diskWriteEnd.Offset()-BackendOffset(d.diskWriteEnd.EndOffset.Pos) {
		nsqLog.Logf("rollback write position can not across file %v, %v, %v", offset, d.diskWriteEnd.Offset(), d.diskWriteEnd.EndOffset.Pos)
		return d.diskWriteEnd, ErrInvalidOffset
	}
	nsqLog.Logf("rollback from %v-%v, %v to %v, roll cnt: %v", d.diskWriteEnd.EndOffset.FileNum, d.diskWriteEnd.EndOffset.Pos, d.diskWriteEnd.Offset(), offset, diffCnt)
	d.diskWriteEnd.EndOffset.Pos -= int64(d.diskWriteEnd.Offset() - offset)
	d.diskWriteEnd.virtualEnd = offset
	atomic.AddInt64(&d.diskWriteEnd.totalMsgCnt, -1*int64(diffCnt))
	if d.diskReadEnd.EndOffset.Pos > d.diskWriteEnd.EndOffset.Pos ||
		d.diskReadEnd.Offset() > d.diskWriteEnd.Offset() {
		d.diskReadEnd = d.diskWriteEnd
	}
	nsqLog.Logf("after rollback : %v, %v, read end: %v", d.diskWriteEnd.EndOffset.Pos, d.diskWriteEnd.TotalMsgCnt(), d.diskReadEnd)
	d.truncateDiskQueueToWriteEnd()
	return d.diskWriteEnd, nil
}
func (d *diskQueueWriter) RollbackWrite(offset BackendOffset, diffCnt uint64) error {
	_, err := d.RollbackWriteV2(offset, diffCnt)
	return err
}
func (d *diskQueueWriter) ResetWriteEnd(offset BackendOffset, totalCnt int64) error {
	_, err := d.ResetWriteEndV2(offset, totalCnt)
	return err
}
func (d *diskQueueWriter) ResetWriteWithQueueStart(queueStart BackendQueueEnd) error {
	d.Lock()
	defer d.Unlock()
	nsqLog.Warningf("DISKQUEUE %v reset the queue start from %v:%v to new queue start: %v", d.name,
		d.diskQueueStart, d.diskWriteEnd, queueStart)
	d.cleanOldData()
	if queueStart.Offset() == 0 && d.diskQueueStart.Offset() == 0 {
		d.diskWriteEnd.EndOffset.FileNum = 0
		d.diskWriteEnd.EndOffset.Pos = 0
	}
	d.diskQueueStart = d.diskWriteEnd
	d.diskQueueStart.virtualEnd = queueStart.Offset()
	d.diskQueueStart.totalMsgCnt = queueStart.TotalMsgCnt()
	d.diskWriteEnd = d.diskQueueStart
	d.diskReadEnd = d.diskWriteEnd
	nsqLog.Warningf("DISKQUEUE %v new queue start : %v:%v", d.name,
		d.diskQueueStart, d.diskWriteEnd)
	d.saveExtraMeta()
	d.persistMetaData(true, d.diskWriteEnd)
	return nil
}
func (d *diskQueueWriter) prepareCleanByRetention(cleanEndInfo BackendQueueOffset,
	noRealClean bool, maxCleanOffset BackendOffset) (BackendQueueEnd, int64, int64, error) {
	if cleanEndInfo == nil {
		return nil, 0, 0, nil
	}
	d.Lock()
	defer d.Unlock()
	newStart := d.diskQueueStart
	var endInfo *diskQueueOffset
	switch e := cleanEndInfo.(type) {
	case *diskQueueOffsetInfo:
		endInfo = &e.EndOffset
	case *diskQueueEndInfo:
		endInfo = &e.EndOffset
	}
	cleanOffset := cleanEndInfo.Offset()
	if endInfo != nil {
		if endInfo.FileNum >= d.diskReadEnd.EndOffset.FileNum-1 {
			endInfo.FileNum = d.diskReadEnd.EndOffset.FileNum - 1
		}
		if endInfo.FileNum <= d.diskQueueStart.EndOffset.FileNum {
			return nil, 0, 0, nil
		}
		if maxCleanOffset != BackendOffset(0) && cleanOffset > maxCleanOffset {
			nsqLog.LogWarningf("disk %v clean position %v exceed the max allowed clean end: %v", d.name, cleanOffset, maxCleanOffset)
			return nil, 0, 0, nil
		}
	} else {
		if cleanOffset >= d.diskReadEnd.Offset()-BackendOffset(d.diskReadEnd.EndOffset.Pos) {
			cleanOffset = d.diskReadEnd.Offset() - BackendOffset(d.diskReadEnd.EndOffset.Pos)
		}
		if maxCleanOffset != BackendOffset(0) && cleanOffset > maxCleanOffset {
			cleanOffset = maxCleanOffset
		}
	}
	cleanFileNum := int64(0)
	if endInfo != nil {
		cleanFileNum = endInfo.FileNum
		if cleanFileNum <= d.diskQueueStart.EndOffset.FileNum {
			return &newStart, 0, 0, nil
		}
	retryClean:
		cnt, _, endPos, err := getQueueFileOffsetMeta(d.fileName(cleanFileNum - 1))
		if err != nil {
			nsqLog.Logf("disk %v failed to get queue offset meta: %v", d.fileName(cleanFileNum), err)
			if os.IsNotExist(err) && cleanFileNum < d.diskReadEnd.EndOffset.FileNum-1 {
				cleanFileNum++
				goto retryClean
			}
			return &newStart, 0, 0, err
		}
		if maxCleanOffset != BackendOffset(0) && BackendOffset(endPos) > maxCleanOffset {
			nsqLog.LogWarningf("disk %v clean position %v exceed the max allowed clean end: %v", d.name, endPos, maxCleanOffset)
			return &newStart, 0, 0, errors.New("clean exceed the max allowed")
		}
		newStart.EndOffset.FileNum = cleanFileNum
		newStart.EndOffset.Pos = 0
		newStart.virtualEnd = BackendOffset(endPos)
		newStart.totalMsgCnt = cnt
	} else if cleanOffset >= 0 {
		checkedNum := newStart.EndOffset.FileNum
		for {
			cnt, _, endPos, err := getQueueFileOffsetMeta(d.fileName(checkedNum))
			if err != nil {
				nsqLog.LogWarningf("disk %v failed to get queue offset meta: %v", checkedNum, err)
				if os.IsNotExist(err) && checkedNum < d.diskReadEnd.EndOffset.FileNum-1 {
					checkedNum++
					continue
				}
				return &newStart, 0, 0, err
			}
			if BackendOffset(endPos) < cleanOffset {
				if checkedNum >= d.diskReadEnd.EndOffset.FileNum-1 {
					break
				}
				checkedNum++
				cleanFileNum = checkedNum
				newStart.EndOffset.FileNum = checkedNum
				newStart.EndOffset.Pos = 0
				newStart.virtualEnd = BackendOffset(endPos)
				newStart.totalMsgCnt = cnt
			} else {
				break
			}
		}
	}
	if noRealClean {
		return &newStart, 0, 0, nil
	}
	nsqLog.Infof("DISKQUEUE %v clean queue from %v, %v to new start : %v", d.name,
		d.diskQueueStart, d.diskWriteEnd, newStart)
	cleanStartFileNum := d.diskQueueStart.EndOffset.FileNum - MAX_QUEUE_OFFSET_META_DATA_KEEP - 1
	if cleanStartFileNum < 0 {
		cleanStartFileNum = 0
	}
	d.diskQueueStart = newStart
	d.saveExtraMeta()
	return &newStart, cleanStartFileNum, cleanFileNum, nil
}
func (d *diskQueueWriter) CleanOldDataByRetention(cleanEndInfo BackendQueueOffset,
	noRealClean bool, maxCleanOffset BackendOffset) (BackendQueueEnd, error) {
	newStart, cleanStartFileNum, cleanFileNum, err := d.prepareCleanByRetention(cleanEndInfo, noRealClean, maxCleanOffset)
	if err != nil {
		return nil, err
	}
	cleanMetaFileNum := cleanFileNum - MAX_QUEUE_OFFSET_META_DATA_KEEP
	for i := cleanStartFileNum; i < cleanFileNum; i++ {
		fn := d.fileName(i)
		innerErr := os.Remove(fn)
		if innerErr != nil {
			if !os.IsNotExist(innerErr) {
				nsqLog.LogErrorf("diskqueue(%s) failed to remove data file %v - %s", d.name, fn, innerErr)
				continue
			}
		} else {
			nsqLog.Logf("DISKQUEUE(%s): removed data file: %v", d.name, fn)
		}
		if i <= cleanMetaFileNum {
			fn = d.fileName(i) + ".offsetmeta.dat"
			innerErr = os.Remove(fn)
			if innerErr != nil {
				if !os.IsNotExist(innerErr) {
					nsqLog.LogErrorf("diskqueue(%s) failed to remove offset meta data file %v - %s", d.name, fn, innerErr)
				}
			} else {
				nsqLog.Debugf("DISKQUEUE(%s): removed offset meta data file: %v", d.name, fn)
			}
		}
	}
	return newStart, nil
}
func (d *diskQueueWriter) closeCurrentFile() {
	if d.bufferWriter != nil {
		d.bufferWriter.Flush()
	}
	if d.diskReadEnd.EndOffset.GreatThan(&d.diskWriteEnd.EndOffset) {
		nsqLog.Logf("DISKQUEUE(%s): old read is greater: %v, %v", d.name,
			d.diskReadEnd, d.diskWriteEnd)
	}
	d.diskReadEnd = d.diskWriteEnd
	if d.writeFile != nil {
		d.writeFile.Close()
		d.writeFile = nil
	}
}
func (d *diskQueueWriter) truncateDiskQueueToWriteEnd() {
	if d.writeFile != nil {
		d.writeFile.Truncate(d.diskWriteEnd.EndOffset.Pos)
		d.writeFile.Close()
		d.writeFile = nil
	} else {
		curFileName := d.fileName(d.diskWriteEnd.EndOffset.FileNum)
		tmpFile, err := os.OpenFile(curFileName, os.O_RDWR|os.O_CREATE, 0644)
		if err != nil {
			nsqLog.LogErrorf("open write queue failed: %v", err)
		} else {
			tmpFile.Truncate(d.diskWriteEnd.EndOffset.Pos)
			tmpFile.Close()
		}
	}
	d.persistMetaData(true, d.diskWriteEnd)
	cleanNum := d.diskWriteEnd.EndOffset.FileNum + 1
	for {
		fileName := d.fileName(cleanNum)
		err := os.Rename(fileName, fileName+".rolldata")
		if err != nil {
			if os.IsNotExist(err) {
				return
			}
			nsqLog.LogErrorf("truncate and remove the write file %v failed: %v", fileName, err)
		}
		nsqLog.LogWarningf("truncate queue and remove the write file %v ", fileName)
		cleanNum++
	}
}