<commit_before>// Copyright 2016 Apcera Inc. All rights reserved.

package stores

import (
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"time"

	"bufio"
	"github.com/nats-io/go-stan/pb"
	"github.com/nats-io/stan-server/spb"
	"github.com/nats-io/stan-server/util"
)

const (
	// Our file version.
	fileVersion = 1

	// Number of files for a MsgStore on a given channel.
	numFiles = 5

	// Name of the subscriptions file.
	subsFileName = "subs.dat"

	// Name of the clients file.
	clientsFileName = "clients.dat"

	// Name of the server file.
	serverFileName = "server.dat"

	// record header size of a subscription:
	// 4 bytes: 1 byte for type, 3 bytes for buffer size
	subRecordHeaderSize = 4
)

// Type for the records in the subscriptions file
type subRecordType byte

// Subscription protobuf do not share a common interface, yet, when saving a
// subscription on disk, we have to get the size and marshal the record in
// a buffer. These methods are available in all the protobuf.
// So we create this interface with those two methods to be used by the
// writeRecord method.
type subRecord interface {
	Size() int
	MarshalTo([]byte) (int, error)
}

// Various record types
const (
	subRecNew = subRecordType(iota)
	subRecUpdate
	subRecDel
	subRecAck
	subRecMsg
)

// Actions for client store
const (
	addClient = "A"
	delClient = "D"
)

// FileStore is the storage interface for STAN servers, backed by files.
type FileStore struct {
	genericStore
	rootDir     string
	serverFile  *os.File
	clientsFile *os.File
}

type recoveredSub struct {
	sub    *spb.SubState
	seqnos map[uint64]struct{}
}

// FileSubStore is a subscription store in files.
type FileSubStore struct {
	genericSubStore
	tmpSubBuf []byte
	file      *os.File
}

// fileSlice represents one of the message store file (there are a number
// of files for a MsgStore on a given channel).
type fileSlice struct {
	fileName  string
	firstMsg  *pb.MsgProto
	lastMsg   *pb.MsgProto
	msgsCount int
	msgsSize  uint64
}

// FileMsgStore is a per channel message file store.
type FileMsgStore struct {
	genericMsgStore
	tmpMsgBuf    []byte
	file         *os.File
	files        [numFiles]*fileSlice
	currSliceIdx int
}

// openFile opens the file specified by `filename`.
// If the file exists, it checks that the version is supported.
// If no file mode is provided, the file is created if not present,
// opened in Read/Write and Append mode.
func openFile(fileName string, modes ...int) (*os.File, error) {
	checkVersion := false

	mode := os.O_RDWR | os.O_CREATE | os.O_APPEND
	if len(modes) > 0 {
		// Use the provided modes instead
		mode = 0
		for _, m := range modes {
			mode |= m
		}
	}

	// Check if file already exists
	if s, err := os.Stat(fileName); s != nil && err == nil {
		checkVersion = true
	}
	file, err := os.OpenFile(fileName, mode, 0666)
	if err != nil {
		return nil, err
	}

	if checkVersion {
		err = checkFileVersion(file)
	} else {
		// This is a new file, write our file version
		err = util.WriteInt(file, fileVersion)
	}
	if err != nil {
		file.Close()
		file = nil
	}
	return file, err
}

// check that the version of the file is understood by this interface
func checkFileVersion(r io.Reader) error {
	fv, err := util.ReadInt(r)
	if err != nil {
		return fmt.Errorf("unable to verify file version: %v", err)
	}
	if fv == 0 || fv > fileVersion {
		return fmt.Errorf("unsupported file version: %v (supports [1..%v])", fv, fileVersion)
	}
	return nil
}

// write the header in the given buffer.
func writeHeader(buf []byte, recType subRecordType, recordSize int) {
	if recordSize > 0xFFFFFF {
		panic("record size too big")
	}
	header := int(recType)<<24 | recordSize
	util.ByteOrder.PutUint32(buf, uint32(header))
}

////////////////////////////////////////////////////////////////////////////
// FileStore methods
////////////////////////////////////////////////////////////////////////////

// NewFileStore returns a factory for stores backed by files, and recovers
// any state present.
// If not limits are provided, the store will be created with
// DefaultChannelLimits.
func NewFileStore(rootDir string, limits *ChannelLimits) (*FileStore, *RecoveredState, error) {
	fs := &FileStore{
		rootDir: rootDir,
	}
	fs.init(TypeFile, limits)

	if err := os.MkdirAll(rootDir, os.ModeDir+os.ModePerm); err != nil && !os.IsExist(err) {
		return nil, nil, fmt.Errorf("unable to create the root directory [%s]: %v", rootDir, err)
	}

	var err error
	var recoveredState *RecoveredState
	var serverInfo *spb.ServerInfo
	var recoveredClients []*RecoveredClient
	var recoveredSubs = make(RecoveredSubscriptions)
	var channels []os.FileInfo
	var msgStore *FileMsgStore
	var subStore *FileSubStore
	var subs map[uint64]*recoveredSub

	// Ensure store is closed in case of return with error
	defer func() {
		if err != nil {
			fs.Close()
		}
	}()

	// Open/Create the server file (note that this file must not be opened,
	// in APPEND mode to allow truncate to work).
	fileName := filepath.Join(fs.rootDir, serverFileName)
	fs.serverFile, err = openFile(fileName, os.O_RDWR, os.O_CREATE)
	if err != nil {
		return nil, nil, err
	}

	// Open/Create the client file.
	fileName = filepath.Join(fs.rootDir, clientsFileName)
	fs.clientsFile, err = openFile(fileName)
	if err != nil {
		return nil, nil, err
	}

	// Recover the server file.
	serverInfo, err = fs.recoverServerInfo()
	if err != nil {
		return nil, nil, err
	}
	// If the server file is empty, then we are done
	if serverInfo == nil {
		// We return the file store instance, but no recovered state.
		return fs, nil, nil
	}

	// Recover the clients file
	recoveredClients, err = fs.recoverClients()
	if err != nil {
		return nil, nil, err
	}

	// Get the channels (there are subdirectories of rootDir)
	channels, err = ioutil.ReadDir(rootDir)
	if err != nil {
		return nil, nil, err
	}
	// Go through the list
	for _, c := range channels {
		// Channels are directories. Ignore simple files
		if !c.IsDir() {
			continue
		}

		channel := c.Name()
		channelDirName := filepath.Join(rootDir, channel)

		// Recover messages for this channel
		msgStore, err = newFileMsgStore(channelDirName, channel, fs.limits, true)
		if err != nil {
			break
		}
		subStore, subs, err = newFileSubStore(channelDirName, channel, fs.limits, true)
		if err != nil {
			msgStore.Close()
			break
		}

		// For this channel, construct an array of RecoveredSubState
		rssArray := make([]*RecoveredSubState, 0, len(subs))

		// Fill that array with what we got from newFileSubStore.
		for _, sub := range subs {
			rss := &RecoveredSubState{
				Sub:     sub.sub,
				Pending: make(PendingAcks),
			}
			// If we recovered any seqno...
			if len(sub.seqnos) > 0 {
				// Lookup messages, and if we find those, update the
				// Pending map.
				for seq := range sub.seqnos {
					// Access directly 'msgs' here. If we have a
					// different implementation where we don't
					// keep messages around, we would still have
					// a cache of messages per channel that will
					// then be cleared after this loop when we
					// are done restoring the subscriptions.
					if m := msgStore.msgs[seq]; m != nil {
						rss.Pending[seq] = m
					}
				}
			}
			// Add to the array of recovered subscriptions
			rssArray = append(rssArray, rss)
		}

		// This is the recovered subscription state for this channel
		recoveredSubs[channel] = rssArray

		fs.channels[channel] = &ChannelStore{
			Subs: subStore,
			Msgs: msgStore,
		}
	}
	if err != nil {
		return nil, nil, err
	}
	// Create the recovered state to return
	recoveredState = &RecoveredState{
		Info:    serverInfo,
		Clients: recoveredClients,
		Subs:    recoveredSubs,
	}
	return fs, recoveredState, nil
}

// Init is used to persist server's information after the first start
func (fs *FileStore) Init(info *spb.ServerInfo) error {
	fs.Lock()
	defer fs.Unlock()

	f := fs.serverFile
	// Truncate the file (4 is the size of the fileVersion record)
	err := f.Truncate(4)
	if err == nil {
		// Move offset to 4 (truncate does not do that)
		_, err = f.Seek(4, 0)
	}
	if err == nil {
		// Write the size of the record first
		err = util.WriteInt(f, info.Size())
	}
	if err == nil {
		b, _ := info.Marshal()
		// Now the content
		_, err = f.Write(b)
	}
	return err
}

// recoverClients reads the client files and returns an array of RecoveredClient
func (fs *FileStore) recoverClients() ([]*RecoveredClient, error) {
	clientsMap := make(map[string]*RecoveredClient)

	var err error
	var action, clientID, hbInbox string

	scanner := bufio.NewScanner(fs.clientsFile)
	for scanner.Scan() {
		line := scanner.Text()
		fmt.Sscanf(line, "%s %s %s", &action, &clientID, &hbInbox)
		switch action {
		case addClient:
			if clientID == "" {
				return nil, fmt.Errorf("missing client ID in ADD client instruction")
			}
			if hbInbox == "" {
				return nil, fmt.Errorf("missing client heartbeat inbox in ADD client instruction")
			}
			c := &RecoveredClient{ClientID: clientID, HbInbox: hbInbox}
			// Add to the map. Note that if one already exists, which should
			// not, just replace with this most recent one.
			clientsMap[clientID] = c
		case delClient:
			if clientID == "" {
				return nil, fmt.Errorf("missing client ID in DELETE client instruction")
			}
			delete(clientsMap, clientID)
		default:
			return nil, fmt.Errorf("invalid client action %q", action)
		}
	}
	err = scanner.Err()
	if err != nil {
		return nil, err
	}
	clients := make([]*RecoveredClient, len(clientsMap))
	i := 0
	// Convert the map into an array
	for _, c := range clientsMap {
		clients[i] = c
		i++
	}
	return clients, nil
}

// recoverServerInfo reads the server file and returns a ServerInfo structure
func (fs *FileStore) recoverServerInfo() (*spb.ServerInfo, error) {
	file := fs.serverFile
	info := &spb.ServerInfo{}
	// Read the message size as an int32
	size, err := util.ReadInt(file)
	if err != nil {
		if err == io.EOF {
			// We are done, no state recovered
			return nil, nil
		}
		return nil, err
	}
	// Check that the size of the file is consistent with the size
	// of the record we are supposed to recover. Account for the
	// 8 bytes (4 + 4) corresponding to the fileVersion and record
	// size.
	fstat, err := file.Stat()
	if err != nil {
		return nil, err
	}
	expectedSize := int64(size + 8)
	if fstat.Size() != expectedSize {
		return nil, fmt.Errorf("incorrect file size, expected %v bytes, got %v bytes",
			expectedSize, fstat.Size())
	}
	buf := make([]byte, size)
	// Read fully the expected number of bytes for this record
	if _, err := io.ReadFull(file, buf); err != nil {
		return nil, err
	}
	// Reconstruct now
	if err := info.Unmarshal(buf); err != nil {
		return nil, err
	}
	return info, nil
}

// LookupOrCreateChannel returns a ChannelStore for the given channel,
// creates one if no such channel exists. In this case, the returned
// boolean will be true.
func (fs *FileStore) LookupOrCreateChannel(channel string) (*ChannelStore, bool, error) {
	channelStore := fs.LookupChannel(channel)
	if channelStore != nil {
		return channelStore, false, nil
	}

	// Two "threads" could make this call and end-up deciding to create the
	// channel. So we need to test again, this time under the write lock.
	fs.Lock()
	defer fs.Unlock()
	channelStore = fs.channels[channel]
	if channelStore != nil {
		return channelStore, false, nil
	}

	// Check for limits
	if err := fs.canAddChannel(); err != nil {
		return nil, false, err
	}

	// We create the channel here...

	channelDirName := filepath.Join(fs.rootDir, channel)
	if err := os.MkdirAll(channelDirName, os.ModeDir+os.ModePerm); err != nil {
		return nil, false, err
	}

	var err error
	var msgStore MsgStore
	var subStore SubStore

	msgStore, err = newFileMsgStore(channelDirName, channel, fs.limits, false)
	if err != nil {
		return nil, false, err
	}
	subStore, _, err = newFileSubStore(channelDirName, channel, fs.limits, false)
	if err != nil {
		msgStore.Close()
		return nil, false, err
	}

	channelStore = &ChannelStore{
		Subs: subStore,
		Msgs: msgStore,
	}

	fs.channels[channel] = channelStore

	return channelStore, true, nil
}

// AddClient stores information about the client identified by `clientID`.
func (fs *FileStore) AddClient(clientID, hbInbox string) error {
	line := fmt.Sprintf("%s %s %s\r\n", addClient, clientID, hbInbox)
	fs.Lock()
	defer fs.Unlock()
	if _, err := fs.clientsFile.WriteString(line); err != nil {
		return err
	}
	return nil
}

// DeleteClient invalidates the client identified by `clientID`.
func (fs *FileStore) DeleteClient(clientID string) {
	line := fmt.Sprintf("%s %s\r\n", delClient, clientID)
	fs.Lock()
	defer fs.Unlock()
	fs.clientsFile.WriteString(line)
}

// Close closes all stores.
func (fs *FileStore) Close() error {
	fs.Lock()
	defer fs.Unlock()
	if fs.closed {
		return nil
	}
	fs.closed = true

	var err error
	closeFile := func(f *os.File) {
		if f == nil {
			return
		}
		if lerr := f.Close(); lerr != nil && err == nil {
			err = lerr
		}
	}
	err = fs.genericStore.close()
	closeFile(fs.serverFile)
	closeFile(fs.clientsFile)
	return err
}

////////////////////////////////////////////////////////////////////////////
// FileMsgStore methods
////////////////////////////////////////////////////////////////////////////

// newFileMsgStore returns a new instace of a file MsgStore.
func newFileMsgStore(channelDirName, channel string, limits ChannelLimits, doRecover bool) (*FileMsgStore, error) {
	var err error
	var file *os.File

	// Create an instance and initialize
	ms := &FileMsgStore{}
	ms.init(channel, limits)

	// Open/create all the files
	for i := 0; i < numFiles; i++ {
		// Fully qualified file name.
		fileName := filepath.Join(channelDirName, fmt.Sprintf("msgs.%d.dat", (i+1)))

		// Open the file.
		file, err = openFile(fileName)
		if err != nil {
			break
		}
		// Save slice
		ms.files[i] = &fileSlice{fileName: fileName}

		// Should we try to recover (startup case)
		if doRecover {
			err = ms.recoverOneMsgFile(file, i)
		} else if i == 0 {
			// Otherwise, keep the first file opened...
			ms.file = file
		} else {
			// and close all others.
			err = file.Close()
		}
		if err != nil {
			break
		}
	}
	// Cleanup on error
	if err != nil {
		ms.Close()
		ms = nil
		action := "create"
		if doRecover {
			action = "recover"
		}
		err = fmt.Errorf("unable to %s message store for [%s]: %v", action, channel, err)
		return nil, err
	}
	return ms, nil
}

// recovers one of the file
func (ms *FileMsgStore) recoverOneMsgFile(file *os.File, numFile int) error {
	var err error

	msgSize := 0
	var msg *pb.MsgProto

	fslice := ms.files[numFile]

	for {
		// Read the message size as an int32
		msgSize, err = util.ReadInt(file)
		if err != nil {
			if err == io.EOF {
				// We are done, reset err
				err = nil
				break
			} else {
				return err
			}
		}

		// Expand buffer if necessary
		ms.tmpMsgBuf = util.EnsureBufBigEnough(ms.tmpMsgBuf, msgSize)

		// Read fully the expected number of bytes for this message
		_, err = io.ReadFull(file, ms.tmpMsgBuf[:msgSize])
		if err != nil {
			return err
		}
		// Recover this message
		msg = &pb.MsgProto{}
		err = msg.Unmarshal(ms.tmpMsgBuf[:msgSize])
		if err != nil {
			return err
		}

		if fslice.firstMsg == nil {
			fslice.firstMsg = msg
		}
		fslice.lastMsg = msg
		fslice.msgsCount++
		fslice.msgsSize += uint64(len(msg.Data))

		if ms.first == 0 {
			ms.first = msg.Sequence
		}
		ms.msgs[msg.Sequence] = msg
	}

	// Do more accounting and bump the current slice index if we recovered
	// at least one message on that file.
	if fslice.msgsCount > 0 {
		ms.last = fslice.lastMsg.Sequence
		ms.totalCount += fslice.msgsCount
		ms.totalBytes += fslice.msgsSize
		ms.currSliceIdx = numFile

		// Close the previous file
		if numFile > 0 {
			err = ms.file.Close()
			ms.file = nil
		}
	}
	// Keep the file opened if this is the first or messages were recovered
	if err == nil && (fslice.msgsCount > 0 || numFile == 0) {
		ms.file = file
	} else {
		// Close otherwise...
		err = file.Close()
	}
	return err
}

// Store a given message.
func (ms *FileMsgStore) Store(reply string, data []byte) (*pb.MsgProto, error) {
	ms.Lock()
	defer ms.Unlock()

	fslice := ms.files[ms.currSliceIdx]

	// Check if we need to move to next file slice
	if (ms.currSliceIdx < numFiles-1) &&
		((fslice.msgsCount >= ms.limits.MaxNumMsgs/(numFiles-1)) ||
			(fslice.msgsSize >= ms.limits.MaxMsgBytes/(numFiles-1))) {

		// Don't change store variable until success...
		nextSlice := ms.currSliceIdx + 1

		// Close the file and open the next slice
		if err := ms.file.Close(); err != nil {
			return nil, err
		}<commit_msg>Fix error handling when recovering message file<commit_after>// Copyright 2016 Apcera Inc. All rights reserved.

package stores

import (
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"time"

	"bufio"
	"github.com/nats-io/go-stan/pb"
	"github.com/nats-io/stan-server/spb"
	"github.com/nats-io/stan-server/util"
)

const (
	// Our file version.
	fileVersion = 1

	// Number of files for a MsgStore on a given channel.
	numFiles = 5

	// Name of the subscriptions file.
	subsFileName = "subs.dat"

	// Name of the clients file.
	clientsFileName = "clients.dat"

	// Name of the server file.
	serverFileName = "server.dat"

	// record header size of a subscription:
	// 4 bytes: 1 byte for type, 3 bytes for buffer size
	subRecordHeaderSize = 4
)

// Type for the records in the subscriptions file
type subRecordType byte

// Subscription protobuf do not share a common interface, yet, when saving a
// subscription on disk, we have to get the size and marshal the record in
// a buffer. These methods are available in all the protobuf.
// So we create this interface with those two methods to be used by the
// writeRecord method.
type subRecord interface {
	Size() int
	MarshalTo([]byte) (int, error)
}

// Various record types
const (
	subRecNew = subRecordType(iota)
	subRecUpdate
	subRecDel
	subRecAck
	subRecMsg
)

// Actions for client store
const (
	addClient = "A"
	delClient = "D"
)

// FileStore is the storage interface for STAN servers, backed by files.
type FileStore struct {
	genericStore
	rootDir     string
	serverFile  *os.File
	clientsFile *os.File
}

type recoveredSub struct {
	sub    *spb.SubState
	seqnos map[uint64]struct{}
}

// FileSubStore is a subscription store in files.
type FileSubStore struct {
	genericSubStore
	tmpSubBuf []byte
	file      *os.File
}

// fileSlice represents one of the message store file (there are a number
// of files for a MsgStore on a given channel).
type fileSlice struct {
	fileName  string
	firstMsg  *pb.MsgProto
	lastMsg   *pb.MsgProto
	msgsCount int
	msgsSize  uint64
}

// FileMsgStore is a per channel message file store.
type FileMsgStore struct {
	genericMsgStore
	tmpMsgBuf    []byte
	file         *os.File
	files        [numFiles]*fileSlice
	currSliceIdx int
}

// openFile opens the file specified by `filename`.
// If the file exists, it checks that the version is supported.
// If no file mode is provided, the file is created if not present,
// opened in Read/Write and Append mode.
func openFile(fileName string, modes ...int) (*os.File, error) {
	checkVersion := false

	mode := os.O_RDWR | os.O_CREATE | os.O_APPEND
	if len(modes) > 0 {
		// Use the provided modes instead
		mode = 0
		for _, m := range modes {
			mode |= m
		}
	}

	// Check if file already exists
	if s, err := os.Stat(fileName); s != nil && err == nil {
		checkVersion = true
	}
	file, err := os.OpenFile(fileName, mode, 0666)
	if err != nil {
		return nil, err
	}

	if checkVersion {
		err = checkFileVersion(file)
	} else {
		// This is a new file, write our file version
		err = util.WriteInt(file, fileVersion)
	}
	if err != nil {
		file.Close()
		file = nil
	}
	return file, err
}

// check that the version of the file is understood by this interface
func checkFileVersion(r io.Reader) error {
	fv, err := util.ReadInt(r)
	if err != nil {
		return fmt.Errorf("unable to verify file version: %v", err)
	}
	if fv == 0 || fv > fileVersion {
		return fmt.Errorf("unsupported file version: %v (supports [1..%v])", fv, fileVersion)
	}
	return nil
}

// write the header in the given buffer.
func writeHeader(buf []byte, recType subRecordType, recordSize int) {
	if recordSize > 0xFFFFFF {
		panic("record size too big")
	}
	header := int(recType)<<24 | recordSize
	util.ByteOrder.PutUint32(buf, uint32(header))
}

////////////////////////////////////////////////////////////////////////////
// FileStore methods
////////////////////////////////////////////////////////////////////////////

// NewFileStore returns a factory for stores backed by files, and recovers
// any state present.
// If not limits are provided, the store will be created with
// DefaultChannelLimits.
func NewFileStore(rootDir string, limits *ChannelLimits) (*FileStore, *RecoveredState, error) {
	fs := &FileStore{
		rootDir: rootDir,
	}
	fs.init(TypeFile, limits)

	if err := os.MkdirAll(rootDir, os.ModeDir+os.ModePerm); err != nil && !os.IsExist(err) {
		return nil, nil, fmt.Errorf("unable to create the root directory [%s]: %v", rootDir, err)
	}

	var err error
	var recoveredState *RecoveredState
	var serverInfo *spb.ServerInfo
	var recoveredClients []*RecoveredClient
	var recoveredSubs = make(RecoveredSubscriptions)
	var channels []os.FileInfo
	var msgStore *FileMsgStore
	var subStore *FileSubStore
	var subs map[uint64]*recoveredSub

	// Ensure store is closed in case of return with error
	defer func() {
		if err != nil {
			fs.Close()
		}
	}()

	// Open/Create the server file (note that this file must not be opened,
	// in APPEND mode to allow truncate to work).
	fileName := filepath.Join(fs.rootDir, serverFileName)
	fs.serverFile, err = openFile(fileName, os.O_RDWR, os.O_CREATE)
	if err != nil {
		return nil, nil, err
	}

	// Open/Create the client file.
	fileName = filepath.Join(fs.rootDir, clientsFileName)
	fs.clientsFile, err = openFile(fileName)
	if err != nil {
		return nil, nil, err
	}

	// Recover the server file.
	serverInfo, err = fs.recoverServerInfo()
	if err != nil {
		return nil, nil, err
	}
	// If the server file is empty, then we are done
	if serverInfo == nil {
		// We return the file store instance, but no recovered state.
		return fs, nil, nil
	}

	// Recover the clients file
	recoveredClients, err = fs.recoverClients()
	if err != nil {
		return nil, nil, err
	}

	// Get the channels (there are subdirectories of rootDir)
	channels, err = ioutil.ReadDir(rootDir)
	if err != nil {
		return nil, nil, err
	}
	// Go through the list
	for _, c := range channels {
		// Channels are directories. Ignore simple files
		if !c.IsDir() {
			continue
		}

		channel := c.Name()
		channelDirName := filepath.Join(rootDir, channel)

		// Recover messages for this channel
		msgStore, err = newFileMsgStore(channelDirName, channel, fs.limits, true)
		if err != nil {
			break
		}
		subStore, subs, err = newFileSubStore(channelDirName, channel, fs.limits, true)
		if err != nil {
			msgStore.Close()
			break
		}

		// For this channel, construct an array of RecoveredSubState
		rssArray := make([]*RecoveredSubState, 0, len(subs))

		// Fill that array with what we got from newFileSubStore.
		for _, sub := range subs {
			rss := &RecoveredSubState{
				Sub:     sub.sub,
				Pending: make(PendingAcks),
			}
			// If we recovered any seqno...
			if len(sub.seqnos) > 0 {
				// Lookup messages, and if we find those, update the
				// Pending map.
				for seq := range sub.seqnos {
					// Access directly 'msgs' here. If we have a
					// different implementation where we don't
					// keep messages around, we would still have
					// a cache of messages per channel that will
					// then be cleared after this loop when we
					// are done restoring the subscriptions.
					if m := msgStore.msgs[seq]; m != nil {
						rss.Pending[seq] = m
					}
				}
			}
			// Add to the array of recovered subscriptions
			rssArray = append(rssArray, rss)
		}

		// This is the recovered subscription state for this channel
		recoveredSubs[channel] = rssArray

		fs.channels[channel] = &ChannelStore{
			Subs: subStore,
			Msgs: msgStore,
		}
	}
	if err != nil {
		return nil, nil, err
	}
	// Create the recovered state to return
	recoveredState = &RecoveredState{
		Info:    serverInfo,
		Clients: recoveredClients,
		Subs:    recoveredSubs,
	}
	return fs, recoveredState, nil
}

// Init is used to persist server's information after the first start
func (fs *FileStore) Init(info *spb.ServerInfo) error {
	fs.Lock()
	defer fs.Unlock()

	f := fs.serverFile
	// Truncate the file (4 is the size of the fileVersion record)
	err := f.Truncate(4)
	if err == nil {
		// Move offset to 4 (truncate does not do that)
		_, err = f.Seek(4, 0)
	}
	if err == nil {
		// Write the size of the record first
		err = util.WriteInt(f, info.Size())
	}
	if err == nil {
		b, _ := info.Marshal()
		// Now the content
		_, err = f.Write(b)
	}
	return err
}

// recoverClients reads the client files and returns an array of RecoveredClient
func (fs *FileStore) recoverClients() ([]*RecoveredClient, error) {
	clientsMap := make(map[string]*RecoveredClient)

	var err error
	var action, clientID, hbInbox string

	scanner := bufio.NewScanner(fs.clientsFile)
	for scanner.Scan() {
		line := scanner.Text()
		fmt.Sscanf(line, "%s %s %s", &action, &clientID, &hbInbox)
		switch action {
		case addClient:
			if clientID == "" {
				return nil, fmt.Errorf("missing client ID in ADD client instruction")
			}
			if hbInbox == "" {
				return nil, fmt.Errorf("missing client heartbeat inbox in ADD client instruction")
			}
			c := &RecoveredClient{ClientID: clientID, HbInbox: hbInbox}
			// Add to the map. Note that if one already exists, which should
			// not, just replace with this most recent one.
			clientsMap[clientID] = c
		case delClient:
			if clientID == "" {
				return nil, fmt.Errorf("missing client ID in DELETE client instruction")
			}
			delete(clientsMap, clientID)
		default:
			return nil, fmt.Errorf("invalid client action %q", action)
		}
	}
	err = scanner.Err()
	if err != nil {
		return nil, err
	}
	clients := make([]*RecoveredClient, len(clientsMap))
	i := 0
	// Convert the map into an array
	for _, c := range clientsMap {
		clients[i] = c
		i++
	}
	return clients, nil
}

// recoverServerInfo reads the server file and returns a ServerInfo structure
func (fs *FileStore) recoverServerInfo() (*spb.ServerInfo, error) {
	file := fs.serverFile
	info := &spb.ServerInfo{}
	// Read the message size as an int32
	size, err := util.ReadInt(file)
	if err != nil {
		if err == io.EOF {
			// We are done, no state recovered
			return nil, nil
		}
		return nil, err
	}
	// Check that the size of the file is consistent with the size
	// of the record we are supposed to recover. Account for the
	// 8 bytes (4 + 4) corresponding to the fileVersion and record
	// size.
	fstat, err := file.Stat()
	if err != nil {
		return nil, err
	}
	expectedSize := int64(size + 8)
	if fstat.Size() != expectedSize {
		return nil, fmt.Errorf("incorrect file size, expected %v bytes, got %v bytes",
			expectedSize, fstat.Size())
	}
	buf := make([]byte, size)
	// Read fully the expected number of bytes for this record
	if _, err := io.ReadFull(file, buf); err != nil {
		return nil, err
	}
	// Reconstruct now
	if err := info.Unmarshal(buf); err != nil {
		return nil, err
	}
	return info, nil
}

// LookupOrCreateChannel returns a ChannelStore for the given channel,
// creates one if no such channel exists. In this case, the returned
// boolean will be true.
func (fs *FileStore) LookupOrCreateChannel(channel string) (*ChannelStore, bool, error) {
	channelStore := fs.LookupChannel(channel)
	if channelStore != nil {
		return channelStore, false, nil
	}

	// Two "threads" could make this call and end-up deciding to create the
	// channel. So we need to test again, this time under the write lock.
	fs.Lock()
	defer fs.Unlock()
	channelStore = fs.channels[channel]
	if channelStore != nil {
		return channelStore, false, nil
	}

	// Check for limits
	if err := fs.canAddChannel(); err != nil {
		return nil, false, err
	}

	// We create the channel here...

	channelDirName := filepath.Join(fs.rootDir, channel)
	if err := os.MkdirAll(channelDirName, os.ModeDir+os.ModePerm); err != nil {
		return nil, false, err
	}

	var err error
	var msgStore MsgStore
	var subStore SubStore

	msgStore, err = newFileMsgStore(channelDirName, channel, fs.limits, false)
	if err != nil {
		return nil, false, err
	}
	subStore, _, err = newFileSubStore(channelDirName, channel, fs.limits, false)
	if err != nil {
		msgStore.Close()
		return nil, false, err
	}

	channelStore = &ChannelStore{
		Subs: subStore,
		Msgs: msgStore,
	}

	fs.channels[channel] = channelStore

	return channelStore, true, nil
}

// AddClient stores information about the client identified by `clientID`.
func (fs *FileStore) AddClient(clientID, hbInbox string) error {
	line := fmt.Sprintf("%s %s %s\r\n", addClient, clientID, hbInbox)
	fs.Lock()
	defer fs.Unlock()
	if _, err := fs.clientsFile.WriteString(line); err != nil {
		return err
	}
	return nil
}

// DeleteClient invalidates the client identified by `clientID`.
func (fs *FileStore) DeleteClient(clientID string) {
	line := fmt.Sprintf("%s %s\r\n", delClient, clientID)
	fs.Lock()
	defer fs.Unlock()
	fs.clientsFile.WriteString(line)
}

// Close closes all stores.
func (fs *FileStore) Close() error {
	fs.Lock()
	defer fs.Unlock()
	if fs.closed {
		return nil
	}
	fs.closed = true

	var err error
	closeFile := func(f *os.File) {
		if f == nil {
			return
		}
		if lerr := f.Close(); lerr != nil && err == nil {
			err = lerr
		}
	}
	err = fs.genericStore.close()
	closeFile(fs.serverFile)
	closeFile(fs.clientsFile)
	return err
}

////////////////////////////////////////////////////////////////////////////
// FileMsgStore methods
////////////////////////////////////////////////////////////////////////////

// newFileMsgStore returns a new instace of a file MsgStore.
func newFileMsgStore(channelDirName, channel string, limits ChannelLimits, doRecover bool) (*FileMsgStore, error) {
	var err error
	var file *os.File

	// Create an instance and initialize
	ms := &FileMsgStore{}
	ms.init(channel, limits)

	// Open/create all the files
	for i := 0; i < numFiles; i++ {
		// Fully qualified file name.
		fileName := filepath.Join(channelDirName, fmt.Sprintf("msgs.%d.dat", (i+1)))

		// Open the file.
		file, err = openFile(fileName)
		if err != nil {
			break
		}
		// Save slice
		ms.files[i] = &fileSlice{fileName: fileName}

		// Should we try to recover (startup case)
		if doRecover {
			err = ms.recoverOneMsgFile(file, i)
		} else if i == 0 {
			// Otherwise, keep the first file opened...
			ms.file = file
		} else {
			// and close all others.
			err = file.Close()
		}
		if err != nil {
			break
		}
	}
	// Cleanup on error
	if err != nil {
		ms.Close()
		ms = nil
		action := "create"
		if doRecover {
			action = "recover"
		}
		err = fmt.Errorf("unable to %s message store for [%s]: %v", action, channel, err)
		return nil, err
	}
	return ms, nil
}

// recovers one of the file
func (ms *FileMsgStore) recoverOneMsgFile(file *os.File, numFile int) error {
	var err error

	msgSize := 0
	var msg *pb.MsgProto

	fslice := ms.files[numFile]

	for {
		// Read the message size as an int32
		msgSize, err = util.ReadInt(file)
		if err != nil {
			if err == io.EOF {
				// We are done, reset err
				err = nil
			}
			break
		}

		// Expand buffer if necessary
		ms.tmpMsgBuf = util.EnsureBufBigEnough(ms.tmpMsgBuf, msgSize)

		// Read fully the expected number of bytes for this message
		_, err = io.ReadFull(file, ms.tmpMsgBuf[:msgSize])
		if err != nil {
			break
		}
		// Recover this message
		msg = &pb.MsgProto{}
		err = msg.Unmarshal(ms.tmpMsgBuf[:msgSize])
		if err != nil {
			break
		}

		if fslice.firstMsg == nil {
			fslice.firstMsg = msg
		}
		fslice.lastMsg = msg
		fslice.msgsCount++
		fslice.msgsSize += uint64(len(msg.Data))

		if ms.first == 0 {
			ms.first = msg.Sequence
		}
		ms.msgs[msg.Sequence] = msg
	}

	// Do more accounting and bump the current slice index if we recovered
	// at least one message on that file.
	if err == nil && fslice.msgsCount > 0 {
		ms.last = fslice.lastMsg.Sequence
		ms.totalCount += fslice.msgsCount
		ms.totalBytes += fslice.msgsSize
		ms.currSliceIdx = numFile

		// Close the previous file
		if numFile > 0 {
			err = ms.file.Close()
			ms.file = nil
		}
	}
	// Keep the file opened if this is the first or messages were recovered
	if err == nil && (fslice.msgsCount > 0 || numFile == 0) {
		ms.file = file
	} else {
		// Close otherwise...
		if lerr := file.Close(); lerr != nil {
			if err == nil {
				err = lerr
			}
		}
	}
	return err
}

// Store a given message.
func (ms *FileMsgStore) Store(reply string, data []byte) (*pb.MsgProto, error) {
	ms.Lock()
	defer ms.Unlock()

	fslice := ms.files[ms.currSliceIdx]

	// Check if we need to move to next file slice
	if (ms.currSliceIdx < numFiles-1) &&
		((fslice.msgsCount >= ms.limits.MaxNumMsgs/(numFiles-1)) ||
			(fslice.msgsSize >= ms.limits.MaxMsgBytes/(numFiles-1))) {

		// Don't change store variable until success...
		nextSlice := ms.currSliceIdx + 1

		// Close the file and open the next slice
		if err := ms.file.Close(); err != nil {
			return nil, err
		}
{'fraction_non_alphanumeric': 0.13035633739287325, 'fraction_numerical': 0.003849045256352428, 'mean_word_length': 2.963294005482064, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 21, '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': '1958855', 'n_tokens_mistral': 11805, 'n_tokens_neox': 10910, 'n_words': 3775}