<commit_before>package wal

import (
	"configuration"
	"fmt"
	"os"
	"path"
	"protocol"
	"sort"
	"strings"
	"sync"
)

type WAL struct {
	config             *configuration.Configuration
	logFiles           []*log
	serverId           uint32
	nextLogFileSuffix  int
	requestsPerLogFile int
	rotateLock         sync.Mutex
}

const HOST_ID_OFFSET = uint64(10000)

func NewWAL(config *configuration.Configuration) (*WAL, error) {
	if config.WalDir == "" {
		return nil, fmt.Errorf("wal directory cannot be empty")
	}
	_, err := os.Stat(config.WalDir)

	if os.IsNotExist(err) {
		err = os.MkdirAll(config.WalDir, 0755)
	}

	if err != nil {
		return nil, err
	}

	dir, err := os.Open(config.WalDir)
	if err != nil {
		return nil, err
	}
	names, err := dir.Readdirnames(-1)
	if err != nil {
		return nil, err
	}

	nextLogFileSuffix := 0
	logFiles := make([]*log, 0)
	for _, name := range names {
		if !strings.HasPrefix(name, "log.") {
			continue
		}
		f, err := os.OpenFile(path.Join(config.WalDir, name), os.O_RDWR, 0644)
		if err != nil {
			return nil, err
		}
		logFile, err := newLog(f, config)
		if err != nil {
			return nil, err
		}
		if suffix := logFile.suffix(); suffix > nextLogFileSuffix {
			nextLogFileSuffix = suffix
		}
		logFiles = append(logFiles, logFile)
	}

	// sort the logfiles by the first request number in the log
	sort.Sort(sortableLogSlice(logFiles))

	wal := &WAL{config: config, logFiles: logFiles, requestsPerLogFile: config.WalRequestsPerLogFile, nextLogFileSuffix: nextLogFileSuffix}

	// if we don't have any log files open yet, open a new one
	if len(logFiles) == 0 {
		_, err = wal.createNewLog()
	}
	return wal, err
}

func (self *WAL) SetServerId(id uint32) {
	self.serverId = id
}

// Marks a given request for a given server as committed
func (self *WAL) Commit(requestNumber uint32, serverId uint32) error {
	lastLogFile := self.logFiles[len(self.logFiles)-1]
	lastLogFile.state.commitRequestNumber(serverId, requestNumber)
	lowestCommitedRequestNumber := lastLogFile.state.LowestCommitedRequestNumber()

	index := self.firstLogFile(lowestCommitedRequestNumber)
	if index == 0 {
		return nil
	}

	var unusedLogFiles []*log
	unusedLogFiles, self.logFiles = self.logFiles[:index], self.logFiles[index:]
	for _, logFile := range unusedLogFiles {
		logFile.close()
		logFile.delete()
	}
	return nil
}

// In the case where this server is running and another one in the cluster stops responding, at some point this server will have to just write
// requests to disk. When the downed server comes back up, it's this server's responsibility to send out any writes that were queued up. If
// the yield function returns nil then the request is committed.
func (self *WAL) RecoverServerFromRequestNumber(requestNumber uint32, shardIds []uint32, yield func(request *protocol.Request, shardId uint32) error) error {
	var firstLogFile int

outer:
	for _, logFile := range self.logFiles[firstLogFile:] {<commit_msg>implement recover from last committed request<commit_after>package wal

import (
	"configuration"
	"fmt"
	"os"
	"path"
	"protocol"
	"sort"
	"strings"
	"sync"
)

type WAL struct {
	config             *configuration.Configuration
	logFiles           []*log
	serverId           uint32
	nextLogFileSuffix  int
	requestsPerLogFile int
	rotateLock         sync.Mutex
}

const HOST_ID_OFFSET = uint64(10000)

func NewWAL(config *configuration.Configuration) (*WAL, error) {
	if config.WalDir == "" {
		return nil, fmt.Errorf("wal directory cannot be empty")
	}
	_, err := os.Stat(config.WalDir)

	if os.IsNotExist(err) {
		err = os.MkdirAll(config.WalDir, 0755)
	}

	if err != nil {
		return nil, err
	}

	dir, err := os.Open(config.WalDir)
	if err != nil {
		return nil, err
	}
	names, err := dir.Readdirnames(-1)
	if err != nil {
		return nil, err
	}

	nextLogFileSuffix := 0
	logFiles := make([]*log, 0)
	for _, name := range names {
		if !strings.HasPrefix(name, "log.") {
			continue
		}
		f, err := os.OpenFile(path.Join(config.WalDir, name), os.O_RDWR, 0644)
		if err != nil {
			return nil, err
		}
		logFile, err := newLog(f, config)
		if err != nil {
			return nil, err
		}
		if suffix := logFile.suffix(); suffix > nextLogFileSuffix {
			nextLogFileSuffix = suffix
		}
		logFiles = append(logFiles, logFile)
	}

	// sort the logfiles by the first request number in the log
	sort.Sort(sortableLogSlice(logFiles))

	wal := &WAL{config: config, logFiles: logFiles, requestsPerLogFile: config.WalRequestsPerLogFile, nextLogFileSuffix: nextLogFileSuffix}

	// if we don't have any log files open yet, open a new one
	if len(logFiles) == 0 {
		_, err = wal.createNewLog()
	}
	return wal, err
}

func (self *WAL) SetServerId(id uint32) {
	self.serverId = id
}

// Marks a given request for a given server as committed
func (self *WAL) Commit(requestNumber uint32, serverId uint32) error {
	lastLogFile := self.logFiles[len(self.logFiles)-1]
	lastLogFile.state.commitRequestNumber(serverId, requestNumber)
	lowestCommitedRequestNumber := lastLogFile.state.LowestCommitedRequestNumber()

	index := self.firstLogFile(lowestCommitedRequestNumber)
	if index == 0 {
		return nil
	}

	var unusedLogFiles []*log
	unusedLogFiles, self.logFiles = self.logFiles[:index], self.logFiles[index:]
	for _, logFile := range unusedLogFiles {
		logFile.close()
		logFile.delete()
	}
	return nil
}

func (self *WAL) RecoverServerFromLastCommit(serverId uint32, shardIds []uint32, yield func(request *protocol.Request, shardId uint32) error) error {
	lastLogFile := self.logFiles[len(self.logFiles)-1]
	requestNumber := lastLogFile.state.ServerLastRequestNumber[serverId]
	return self.RecoverServerFromRequestNumber(requestNumber, shardIds, yield)
}

// In the case where this server is running and another one in the cluster stops responding, at some point this server will have to just write
// requests to disk. When the downed server comes back up, it's this server's responsibility to send out any writes that were queued up. If
// the yield function returns nil then the request is committed.
func (self *WAL) RecoverServerFromRequestNumber(requestNumber uint32, shardIds []uint32, yield func(request *protocol.Request, shardId uint32) error) error {
	var firstLogFile int

outer:
	for _, logFile := range self.logFiles[firstLogFile:] {
{'fraction_non_alphanumeric': 0.10746841516072285, 'fraction_numerical': 0.01231408923716616, 'mean_word_length': 4.088689991863303, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '4283430', 'n_tokens_mistral': 2210, 'n_tokens_neox': 2068, 'n_words': 621}