<commit_before>package nsqd

import (
	"bytes"
	"encoding/binary"
	"errors"
	"github.com/absolute8511/nsq/internal/levellogger"
	"github.com/boltdb/bolt"
	"io"
	"os"
	"path"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
)

var (
	syncedOffsetKey  = []byte("synced_offset")
	bucketDelayedMsg = []byte("delayed_message")
	bucketMeta       = []byte("meta")
	CompactThreshold = 1024 * 1024 * 16
)

const (
	MinDelayedType      = 1
	ChannelDelayed      = 1
	PubDelayed          = 2
	TransactionDelayed  = 3
	MaxDelayedType      = 4
	TxMaxSize           = 65536
	CompactCntThreshold = 20000
)

type RecentKeyList [][]byte

func writeDelayedMessageToBackend(buf *bytes.Buffer, msg *Message, bq *diskQueueWriter,
	isExt bool) (BackendOffset, int32, diskQueueEndInfo, error) {
	buf.Reset()
	_, err := msg.WriteDelayedTo(buf, isExt)
	if err != nil {
		return 0, 0, diskQueueEndInfo{}, err
	}
	return bq.PutV2(buf.Bytes())
}

func IsValidDelayedMessage(m *Message) bool {
	if m.DelayedType == ChannelDelayed {
		return m.DelayedOrigID > 0 && len(m.DelayedChannel) > 0 && m.DelayedTs > 0
	} else if m.DelayedType == PubDelayed {
		return m.DelayedTs > 0
	} else if m.DelayedType == TransactionDelayed {
		return true
	}
	return false
}

func getDelayedMsgDBKey(dt int, ch string, ts int64, id MessageID) []byte {
	msgKey := make([]byte, len(ch)+2+1+2+8+8)
	binary.BigEndian.PutUint16(msgKey[:2], uint16(dt))
	pos := 2
	msgKey[pos] = '-'
	pos++
	binary.BigEndian.PutUint16(msgKey[pos:pos+2], uint16(len(ch)))
	pos += 2
	copy(msgKey[pos:pos+len(ch)], []byte(ch))
	pos += len(ch)
	binary.BigEndian.PutUint64(msgKey[pos:pos+8], uint64(ts))
	pos += 8
	binary.BigEndian.PutUint64(msgKey[pos:pos+8], uint64(id))
	return msgKey
}

func decodeDelayedMsgDBKey(b []byte) (uint16, int64, MessageID, string, error) {
	if len(b) < 2+1+2+8+8 {
		return 0, 0, 0, "", errors.New("invalid buffer length")
	}
	dt := binary.BigEndian.Uint16(b[:2])
	pos := 2
	pos++
	chLen := int(binary.BigEndian.Uint16(b[pos : pos+2]))
	pos += 2
	if len(b) < pos+chLen {
		return 0, 0, 0, "", errors.New("invalid buffer length")
	}
	ch := b[pos : pos+chLen]
	pos += chLen
	ts := int64(binary.BigEndian.Uint64(b[pos : pos+8]))
	pos += 8
	id := int64(binary.BigEndian.Uint64(b[pos : pos+8]))
	return dt, ts, MessageID(id), string(ch), nil
}

func getDelayedMsgDBPrefixKey(dt int, ch string) []byte {
	msgKey := make([]byte, len(ch)+2+1+2)
	binary.BigEndian.PutUint16(msgKey[:2], uint16(dt))
	pos := 2
	msgKey[pos] = '-'
	pos++
	binary.BigEndian.PutUint16(msgKey[pos:pos+2], uint16(len(ch)))
	pos += 2
	copy(msgKey[pos:pos+len(ch)], []byte(ch))
	return msgKey
}

func deleteBucketKey(dt int, ch string, ts int64, id MessageID, tx *bolt.Tx) error {
	b := tx.Bucket(bucketDelayedMsg)
	msgKey := getDelayedMsgDBKey(dt, ch, ts, id)
	oldV := b.Get(msgKey)
	err := b.Delete(msgKey)
	if err != nil {
		nsqLog.Infof("failed to delete delayed message: %v", msgKey)
		return err
	}
	if oldV != nil {
		b = tx.Bucket(bucketMeta)
		cntKey := append([]byte("counter_"), getDelayedMsgDBPrefixKey(dt, ch)...)
		cnt := uint64(0)
		cntBytes := b.Get(cntKey)
		if cntBytes != nil && len(cntBytes) == 8 {
			cnt = binary.BigEndian.Uint64(cntBytes)
		}
		if cnt > 0 {
			cnt--
			cntBytes = make([]byte, 8)
			binary.BigEndian.PutUint64(cntBytes[:8], cnt)
			err = b.Put(cntKey, cntBytes)
			if err != nil {
				nsqLog.Infof("failed to update the meta count: %v, %v", cntKey, err)
				return err
			}
		}
	} else {
		nsqLog.Infof("failed to get the deleting delayed message: %v", msgKey)
		return errors.New("key not found")
	}
	return nil
}

type DelayQueue struct {
	tname     string
	fullName  string
	partition int
	backend   *diskQueueWriter
	dataPath  string
	exitFlag  int32

	msgIDCursor  MsgIDGenerator
	defaultIDSeq uint64

	needFlush   int32
	putBuffer   bytes.Buffer
	kvStore     *bolt.DB
	EnableTrace int32
	SyncEvery   int64
	lastSyncCnt int64
	needFixData int32
	isExt       bool
	dbLock      sync.Mutex
	// prevent write while compact db
	compactMutex           sync.Mutex
	oldestChannelDelayedTs map[string]int64
	oldestMutex            sync.Mutex
}

func NewDelayQueueForRead(topicName string, part int, dataPath string, opt *Options,
	idGen MsgIDGenerator, isExt bool) (*DelayQueue, error) {
	ro := &bolt.Options{
		Timeout:  time.Second,
		ReadOnly: true,
	}
	return newDelayQueue(topicName, part, dataPath, opt, idGen, isExt, ro)
}

func NewDelayQueue(topicName string, part int, dataPath string, opt *Options,
	idGen MsgIDGenerator, isExt bool) (*DelayQueue, error) {

	return newDelayQueue(topicName, part, dataPath, opt, idGen, isExt, nil)
}
func newDelayQueue(topicName string, part int, dataPath string, opt *Options,
	idGen MsgIDGenerator, isExt bool, ro *bolt.Options) (*DelayQueue, error) {
	dataPath = path.Join(dataPath, "delayed_queue")
	os.MkdirAll(dataPath, 0755)
	q := &DelayQueue{
		tname:       topicName,
		partition:   part,
		putBuffer:   bytes.Buffer{},
		dataPath:    dataPath,
		msgIDCursor: idGen,
		isExt:       isExt,
		oldestChannelDelayedTs: make(map[string]int64),
	}
	q.fullName = GetTopicFullName(q.tname, q.partition)
	backendName := getDelayQueueBackendName(q.tname, q.partition)
	// max delay message size need add the delay ts and channel name
	queue, err := NewDiskQueueWriter(backendName,
		q.dataPath,
		opt.MaxBytesPerFile,
		int32(minValidMsgLength),
		int32(opt.MaxMsgSize)+minValidMsgLength+8+255, 0)

	if err != nil {
		nsqLog.LogErrorf("topic(%v) failed to init delayed disk queue: %v , %v ", q.fullName, err, backendName)
		return nil, err
	}
	q.backend = queue.(*diskQueueWriter)
	q.kvStore, err = bolt.Open(path.Join(q.dataPath, getDelayQueueDBName(q.tname, q.partition)), 0644, ro)
	if err != nil {
		nsqLog.LogErrorf("topic(%v) failed to init delayed db: %v , %v ", q.fullName, err, backendName)
		return nil, err
	}
	q.kvStore.NoSync = true
	err = q.kvStore.Update(func(tx *bolt.Tx) error {
		_, err := tx.CreateBucketIfNotExists(bucketDelayedMsg)
		if err != nil {
			return err
		}
		_, err = tx.CreateBucketIfNotExists(bucketMeta)
		return err
	})
	if err != nil {
		nsqLog.LogErrorf("topic(%v) failed to init delayed db: %v , %v ", q.fullName, err, backendName)
		if ro == nil || !ro.ReadOnly {
			return nil, err
		}
	}

	return q, nil
}

func (q *DelayQueue) CheckConsistence() error {
	// Perform consistency check.
	return q.getStore().View(func(tx *bolt.Tx) error {
		var count int
		ch := tx.Check()
		done := false
		for !done {
			select {
			case err, ok := <-ch:
				if !ok {
					done = true
					break
				}
				nsqLog.LogErrorf("topic(%v) failed to check delayed db: %v ", q.fullName, err)
				count++
			}
		}

		if count > 0 {
			nsqLog.LogErrorf("topic(%v) failed to check delayed db, %d errors found ", q.fullName, count)
			return errors.New("boltdb file corrupt")
		}
		return nil
	})
}<commit_msg>ignore check error for unfreed<commit_after>package nsqd

import (
	"bytes"
	"encoding/binary"
	"errors"
	"github.com/absolute8511/nsq/internal/levellogger"
	"github.com/boltdb/bolt"
	"io"
	"os"
	"path"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

var (
	syncedOffsetKey  = []byte("synced_offset")
	bucketDelayedMsg = []byte("delayed_message")
	bucketMeta       = []byte("meta")
	CompactThreshold = 1024 * 1024 * 16
)

const (
	MinDelayedType      = 1
	ChannelDelayed      = 1
	PubDelayed          = 2
	TransactionDelayed  = 3
	MaxDelayedType      = 4
	TxMaxSize           = 65536
	CompactCntThreshold = 20000
)

type RecentKeyList [][]byte

func writeDelayedMessageToBackend(buf *bytes.Buffer, msg *Message, bq *diskQueueWriter,
	isExt bool) (BackendOffset, int32, diskQueueEndInfo, error) {
	buf.Reset()
	_, err := msg.WriteDelayedTo(buf, isExt)
	if err != nil {
		return 0, 0, diskQueueEndInfo{}, err
	}
	return bq.PutV2(buf.Bytes())
}

func IsValidDelayedMessage(m *Message) bool {
	if m.DelayedType == ChannelDelayed {
		return m.DelayedOrigID > 0 && len(m.DelayedChannel) > 0 && m.DelayedTs > 0
	} else if m.DelayedType == PubDelayed {
		return m.DelayedTs > 0
	} else if m.DelayedType == TransactionDelayed {
		return true
	}
	return false
}

func getDelayedMsgDBKey(dt int, ch string, ts int64, id MessageID) []byte {
	msgKey := make([]byte, len(ch)+2+1+2+8+8)
	binary.BigEndian.PutUint16(msgKey[:2], uint16(dt))
	pos := 2
	msgKey[pos] = '-'
	pos++
	binary.BigEndian.PutUint16(msgKey[pos:pos+2], uint16(len(ch)))
	pos += 2
	copy(msgKey[pos:pos+len(ch)], []byte(ch))
	pos += len(ch)
	binary.BigEndian.PutUint64(msgKey[pos:pos+8], uint64(ts))
	pos += 8
	binary.BigEndian.PutUint64(msgKey[pos:pos+8], uint64(id))
	return msgKey
}

func decodeDelayedMsgDBKey(b []byte) (uint16, int64, MessageID, string, error) {
	if len(b) < 2+1+2+8+8 {
		return 0, 0, 0, "", errors.New("invalid buffer length")
	}
	dt := binary.BigEndian.Uint16(b[:2])
	pos := 2
	pos++
	chLen := int(binary.BigEndian.Uint16(b[pos : pos+2]))
	pos += 2
	if len(b) < pos+chLen {
		return 0, 0, 0, "", errors.New("invalid buffer length")
	}
	ch := b[pos : pos+chLen]
	pos += chLen
	ts := int64(binary.BigEndian.Uint64(b[pos : pos+8]))
	pos += 8
	id := int64(binary.BigEndian.Uint64(b[pos : pos+8]))
	return dt, ts, MessageID(id), string(ch), nil
}

func getDelayedMsgDBPrefixKey(dt int, ch string) []byte {
	msgKey := make([]byte, len(ch)+2+1+2)
	binary.BigEndian.PutUint16(msgKey[:2], uint16(dt))
	pos := 2
	msgKey[pos] = '-'
	pos++
	binary.BigEndian.PutUint16(msgKey[pos:pos+2], uint16(len(ch)))
	pos += 2
	copy(msgKey[pos:pos+len(ch)], []byte(ch))
	return msgKey
}

func deleteBucketKey(dt int, ch string, ts int64, id MessageID, tx *bolt.Tx) error {
	b := tx.Bucket(bucketDelayedMsg)
	msgKey := getDelayedMsgDBKey(dt, ch, ts, id)
	oldV := b.Get(msgKey)
	err := b.Delete(msgKey)
	if err != nil {
		nsqLog.Infof("failed to delete delayed message: %v", msgKey)
		return err
	}
	if oldV != nil {
		b = tx.Bucket(bucketMeta)
		cntKey := append([]byte("counter_"), getDelayedMsgDBPrefixKey(dt, ch)...)
		cnt := uint64(0)
		cntBytes := b.Get(cntKey)
		if cntBytes != nil && len(cntBytes) == 8 {
			cnt = binary.BigEndian.Uint64(cntBytes)
		}
		if cnt > 0 {
			cnt--
			cntBytes = make([]byte, 8)
			binary.BigEndian.PutUint64(cntBytes[:8], cnt)
			err = b.Put(cntKey, cntBytes)
			if err != nil {
				nsqLog.Infof("failed to update the meta count: %v, %v", cntKey, err)
				return err
			}
		}
	} else {
		nsqLog.Infof("failed to get the deleting delayed message: %v", msgKey)
		return errors.New("key not found")
	}
	return nil
}

type DelayQueue struct {
	tname     string
	fullName  string
	partition int
	backend   *diskQueueWriter
	dataPath  string
	exitFlag  int32

	msgIDCursor  MsgIDGenerator
	defaultIDSeq uint64

	needFlush   int32
	putBuffer   bytes.Buffer
	kvStore     *bolt.DB
	EnableTrace int32
	SyncEvery   int64
	lastSyncCnt int64
	needFixData int32
	isExt       bool
	dbLock      sync.Mutex
	// prevent write while compact db
	compactMutex           sync.Mutex
	oldestChannelDelayedTs map[string]int64
	oldestMutex            sync.Mutex
}

func NewDelayQueueForRead(topicName string, part int, dataPath string, opt *Options,
	idGen MsgIDGenerator, isExt bool) (*DelayQueue, error) {
	ro := &bolt.Options{
		Timeout:  time.Second,
		ReadOnly: true,
	}
	return newDelayQueue(topicName, part, dataPath, opt, idGen, isExt, ro)
}

func NewDelayQueue(topicName string, part int, dataPath string, opt *Options,
	idGen MsgIDGenerator, isExt bool) (*DelayQueue, error) {

	return newDelayQueue(topicName, part, dataPath, opt, idGen, isExt, nil)
}
func newDelayQueue(topicName string, part int, dataPath string, opt *Options,
	idGen MsgIDGenerator, isExt bool, ro *bolt.Options) (*DelayQueue, error) {
	dataPath = path.Join(dataPath, "delayed_queue")
	os.MkdirAll(dataPath, 0755)
	q := &DelayQueue{
		tname:       topicName,
		partition:   part,
		putBuffer:   bytes.Buffer{},
		dataPath:    dataPath,
		msgIDCursor: idGen,
		isExt:       isExt,
		oldestChannelDelayedTs: make(map[string]int64),
	}
	q.fullName = GetTopicFullName(q.tname, q.partition)
	backendName := getDelayQueueBackendName(q.tname, q.partition)
	// max delay message size need add the delay ts and channel name
	queue, err := NewDiskQueueWriter(backendName,
		q.dataPath,
		opt.MaxBytesPerFile,
		int32(minValidMsgLength),
		int32(opt.MaxMsgSize)+minValidMsgLength+8+255, 0)

	if err != nil {
		nsqLog.LogErrorf("topic(%v) failed to init delayed disk queue: %v , %v ", q.fullName, err, backendName)
		return nil, err
	}
	q.backend = queue.(*diskQueueWriter)
	q.kvStore, err = bolt.Open(path.Join(q.dataPath, getDelayQueueDBName(q.tname, q.partition)), 0644, ro)
	if err != nil {
		nsqLog.LogErrorf("topic(%v) failed to init delayed db: %v , %v ", q.fullName, err, backendName)
		return nil, err
	}
	q.kvStore.NoSync = true
	err = q.kvStore.Update(func(tx *bolt.Tx) error {
		_, err := tx.CreateBucketIfNotExists(bucketDelayedMsg)
		if err != nil {
			return err
		}
		_, err = tx.CreateBucketIfNotExists(bucketMeta)
		return err
	})
	if err != nil {
		nsqLog.LogErrorf("topic(%v) failed to init delayed db: %v , %v ", q.fullName, err, backendName)
		if ro == nil || !ro.ReadOnly {
			return nil, err
		}
	}

	return q, nil
}

func (q *DelayQueue) CheckConsistence() error {
	// Perform consistency check.
	return q.getStore().View(func(tx *bolt.Tx) error {
		var count int
		ch := tx.Check()
		done := false
		for !done {
			select {
			case err, ok := <-ch:
				if !ok {
					done = true
					break
				}
				nsqLog.LogErrorf("topic(%v) failed to check delayed db: %v ", q.fullName, err)
				if err != nil && strings.Contains(err.Error(), "unreachable unfreed") {
					continue
				}
				count++
			}
		}

		if count > 0 {
			nsqLog.LogErrorf("topic(%v) failed to check delayed db, %d errors found ", q.fullName, count)
			return errors.New("boltdb file corrupt")
		}
		return nil
	})
}
{'fraction_non_alphanumeric': 0.13892119802268102, 'fraction_numerical': 0.024134922942715905, 'mean_word_length': 3.9467817331895003, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2054129', 'n_tokens_mistral': 5638, 'n_tokens_neox': 5216, 'n_words': 1258}