rename blockstore to filestore. rename blockid to zoneid.

2026-08-05 13:57:07 -07:00 · 2024-06-03 13:03:21 -07:00
parent f7c2897904
commit c191fc8945
14 changed files with 385 additions and 367 deletions
@@ -0,0 +1,435 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package filestore
+
+// the blockstore package implements a write cache for wave files
+// it is not a read cache (reads still go to the DB -- unless items are in the cache)
+// but all writes only go to the cache, and then the cache is periodically flushed to the DB
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"log"
+	"runtime/debug"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+const DefaultPartDataSize = 64 * 1024
+const DefaultFlushTime = 5 * time.Second
+const NoPartIdx = -1
+
+// for unit tests
+var warningCount = &atomic.Int32{}
+var flushErrorCount = &atomic.Int32{}
+
+var partDataSize int64 = DefaultPartDataSize // overridden in tests
+var stopFlush = &atomic.Bool{}
+
+var WFS *FileStore = &FileStore{
+	Lock:  &sync.Mutex{},
+	Cache: make(map[cacheKey]*CacheEntry),
+}
+
+type FileOptsType struct {
+	MaxSize  int64 `json:"maxsize,omitempty"`
+	Circular bool  `json:"circular,omitempty"`
+	IJson    bool  `json:"ijson,omitempty"`
+}
+
+type FileMeta = map[string]any
+
+type WaveFile struct {
+	// these fields are static (not updated)
+	ZoneId    string       `json:"zoneid"`
+	Name      string       `json:"name"`
+	Opts      FileOptsType `json:"opts"`
+	CreatedTs int64        `json:"createdts"`
+
+	//  these fields are mutable
+	Size  int64    `json:"size"`
+	ModTs int64    `json:"modts"`
+	Meta  FileMeta `json:"meta"` // only top-level keys can be updated (lower levels are immutable)
+}
+
+// for regular files this is just Size
+// for circular files this is min(Size, MaxSize)
+func (f WaveFile) DataLength() int64 {
+	if f.Opts.Circular {
+		return minInt64(f.Size, f.Opts.MaxSize)
+	}
+	return f.Size
+}
+
+// for regular files this is just 0
+// for circular files this is the index of the first byte of data we have
+func (f WaveFile) DataStartIdx() int64 {
+	if f.Opts.Circular && f.Size > f.Opts.MaxSize {
+		return f.Size - f.Opts.MaxSize
+	}
+	return 0
+}
+
+// this works because lower levels are immutable
+func copyMeta(meta FileMeta) FileMeta {
+	newMeta := make(FileMeta)
+	for k, v := range meta {
+		newMeta[k] = v
+	}
+	return newMeta
+}
+
+func (f *WaveFile) DeepCopy() *WaveFile {
+	if f == nil {
+		return nil
+	}
+	newFile := *f
+	newFile.Meta = copyMeta(f.Meta)
+	return &newFile
+}
+
+func (WaveFile) UseDBMap() {}
+
+type FileData struct {
+	ZoneId  string `json:"zoneid"`
+	Name    string `json:"name"`
+	PartIdx int    `json:"partidx"`
+	Data    []byte `json:"data"`
+}
+
+func (FileData) UseDBMap() {}
+
+// synchronous (does not interact with the cache)
+func (s *FileStore) MakeFile(ctx context.Context, zoneId string, name string, meta FileMeta, opts FileOptsType) error {
+	if opts.MaxSize < 0 {
+		return fmt.Errorf("max size must be non-negative")
+	}
+	if opts.Circular && opts.MaxSize <= 0 {
+		return fmt.Errorf("circular file must have a max size")
+	}
+	if opts.Circular && opts.IJson {
+		return fmt.Errorf("circular file cannot be ijson")
+	}
+	if opts.Circular {
+		if opts.MaxSize%partDataSize != 0 {
+			opts.MaxSize = (opts.MaxSize/partDataSize + 1) * partDataSize
+		}
+	}
+	return withLock(s, zoneId, name, func(entry *CacheEntry) error {
+		if entry.File != nil {
+			return fs.ErrExist
+		}
+		now := time.Now().UnixMilli()
+		file := &WaveFile{
+			ZoneId:    zoneId,
+			Name:      name,
+			Size:      0,
+			CreatedTs: now,
+			ModTs:     now,
+			Opts:      opts,
+			Meta:      meta,
+		}
+		return dbInsertFile(ctx, file)
+	})
+}
+
+func (s *FileStore) DeleteFile(ctx context.Context, zoneId string, name string) error {
+	return withLock(s, zoneId, name, func(entry *CacheEntry) error {
+		err := dbDeleteFile(ctx, zoneId, name)
+		if err != nil {
+			return fmt.Errorf("error deleting file: %v", err)
+		}
+		entry.clear()
+		return nil
+	})
+}
+
+func (s *FileStore) DeleteZone(ctx context.Context, zoneId string) error {
+	fileNames, err := dbGetZoneFileNames(ctx, zoneId)
+	if err != nil {
+		return fmt.Errorf("error getting zone files: %v", err)
+	}
+	for _, name := range fileNames {
+		s.DeleteFile(ctx, zoneId, name)
+	}
+	return nil
+}
+
+// if file doesn't exsit, returns fs.ErrNotExist
+func (s *FileStore) Stat(ctx context.Context, zoneId string, name string) (*WaveFile, error) {
+	return withLockRtn(s, zoneId, name, func(entry *CacheEntry) (*WaveFile, error) {
+		file, err := entry.loadFileForRead(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("error getting file: %v", err)
+		}
+		return file.DeepCopy(), nil
+	})
+}
+
+func (s *FileStore) ListFiles(ctx context.Context, zoneId string) ([]*WaveFile, error) {
+	files, err := dbGetZoneFiles(ctx, zoneId)
+	if err != nil {
+		return nil, fmt.Errorf("error getting zone files: %v", err)
+	}
+	for idx, file := range files {
+		withLock(s, file.ZoneId, file.Name, func(entry *CacheEntry) error {
+			if entry.File != nil {
+				files[idx] = entry.File.DeepCopy()
+			}
+			return nil
+		})
+	}
+	return files, nil
+}
+
+func (s *FileStore) WriteMeta(ctx context.Context, zoneId string, name string, meta FileMeta, merge bool) error {
+	return withLock(s, zoneId, name, func(entry *CacheEntry) error {
+		err := entry.loadFileIntoCache(ctx)
+		if err != nil {
+			return err
+		}
+		if merge {
+			for k, v := range meta {
+				if v == nil {
+					delete(entry.File.Meta, k)
+					continue
+				}
+				entry.File.Meta[k] = v
+			}
+		} else {
+			entry.File.Meta = meta
+		}
+		entry.File.ModTs = time.Now().UnixMilli()
+		return nil
+	})
+}
+
+func (s *FileStore) WriteFile(ctx context.Context, zoneId string, name string, data []byte) error {
+	return withLock(s, zoneId, name, func(entry *CacheEntry) error {
+		err := entry.loadFileIntoCache(ctx)
+		if err != nil {
+			return err
+		}
+		entry.writeAt(0, data, true)
+		// since WriteFile can *truncate* the file, we need to flush the file to the DB immediately
+		return entry.flushToDB(ctx, true)
+	})
+}
+
+func (s *FileStore) WriteAt(ctx context.Context, zoneId string, name string, offset int64, data []byte) error {
+	if offset < 0 {
+		return fmt.Errorf("offset must be non-negative")
+	}
+	return withLock(s, zoneId, name, func(entry *CacheEntry) error {
+		err := entry.loadFileIntoCache(ctx)
+		if err != nil {
+			return err
+		}
+		file := entry.File
+		if offset > file.Size {
+			return fmt.Errorf("offset is past the end of the file")
+		}
+		partMap := file.computePartMap(offset, int64(len(data)))
+		incompleteParts := incompletePartsFromMap(partMap)
+		err = entry.loadDataPartsIntoCache(ctx, incompleteParts)
+		if err != nil {
+			return err
+		}
+		entry.writeAt(offset, data, false)
+		return nil
+	})
+}
+
+func (s *FileStore) AppendData(ctx context.Context, zoneId string, name string, data []byte) error {
+	return withLock(s, zoneId, name, func(entry *CacheEntry) error {
+		err := entry.loadFileIntoCache(ctx)
+		if err != nil {
+			return err
+		}
+		partMap := entry.File.computePartMap(entry.File.Size, int64(len(data)))
+		incompleteParts := incompletePartsFromMap(partMap)
+		if len(incompleteParts) > 0 {
+			err = entry.loadDataPartsIntoCache(ctx, incompleteParts)
+			if err != nil {
+				return err
+			}
+		}
+		entry.writeAt(entry.File.Size, data, false)
+		return nil
+	})
+}
+
+func (s *FileStore) GetAllZoneIds(ctx context.Context) ([]string, error) {
+	return dbGetAllZoneIds(ctx)
+}
+
+// returns (offset, data, error)
+// we return the offset because the offset may have been adjusted if the size was too big (for circular files)
+func (s *FileStore) ReadAt(ctx context.Context, zoneId string, name string, offset int64, size int64) (rtnOffset int64, rtnData []byte, rtnErr error) {
+	withLock(s, zoneId, name, func(entry *CacheEntry) error {
+		rtnOffset, rtnData, rtnErr = entry.readAt(ctx, offset, size, false)
+		return nil
+	})
+	return
+}
+
+// returns (offset, data, error)
+func (s *FileStore) ReadFile(ctx context.Context, zoneId string, name string) (rtnOffset int64, rtnData []byte, rtnErr error) {
+	withLock(s, zoneId, name, func(entry *CacheEntry) error {
+		rtnOffset, rtnData, rtnErr = entry.readAt(ctx, 0, 0, true)
+		return nil
+	})
+	return
+}
+
+type FlushStats struct {
+	FlushDuration   time.Duration
+	NumDirtyEntries int
+	NumCommitted    int
+}
+
+func (s *FileStore) FlushCache(ctx context.Context) (stats FlushStats, rtnErr error) {
+	wasFlushing := s.setUnlessFlushing()
+	if wasFlushing {
+		return stats, fmt.Errorf("flush already in progress")
+	}
+	defer s.setIsFlushing(false)
+	startTime := time.Now()
+	defer func() {
+		stats.FlushDuration = time.Since(startTime)
+	}()
+
+	// get a copy of dirty keys so we can iterate without the lock
+	dirtyCacheKeys := s.getDirtyCacheKeys()
+	stats.NumDirtyEntries = len(dirtyCacheKeys)
+	for _, key := range dirtyCacheKeys {
+		err := withLock(s, key.ZoneId, key.Name, func(entry *CacheEntry) error {
+			return entry.flushToDB(ctx, false)
+		})
+		if ctx.Err() != nil {
+			// transient error (also must stop the loop)
+			return stats, ctx.Err()
+		}
+		if err != nil {
+			return stats, fmt.Errorf("error flushing cache entry[%v]: %v", key, err)
+		}
+		stats.NumCommitted++
+	}
+	return stats, nil
+}
+
+///////////////////////////////////
+
+func (f *WaveFile) partIdxAtOffset(offset int64) int {
+	partIdx := int(offset / partDataSize)
+	if f.Opts.Circular {
+		maxPart := int(f.Opts.MaxSize / partDataSize)
+		partIdx = partIdx % maxPart
+	}
+	return partIdx
+}
+
+func incompletePartsFromMap(partMap map[int]int) []int {
+	var incompleteParts []int
+	for partIdx, size := range partMap {
+		if size != int(partDataSize) {
+			incompleteParts = append(incompleteParts, partIdx)
+		}
+	}
+	return incompleteParts
+}
+
+func getPartIdxsFromMap(partMap map[int]int) []int {
+	var partIdxs []int
+	for partIdx := range partMap {
+		partIdxs = append(partIdxs, partIdx)
+	}
+	return partIdxs
+}
+
+// returns a map of partIdx to amount of data to write to that part
+func (file *WaveFile) computePartMap(startOffset int64, size int64) map[int]int {
+	partMap := make(map[int]int)
+	endOffset := startOffset + size
+	startFileOffset := startOffset - (startOffset % partDataSize)
+	for testOffset := startFileOffset; testOffset < endOffset; testOffset += partDataSize {
+		partIdx := file.partIdxAtOffset(testOffset)
+		partStartOffset := testOffset
+		partEndOffset := testOffset + partDataSize
+		partWriteStartOffset := 0
+		partWriteEndOffset := int(partDataSize)
+		if startOffset > partStartOffset && startOffset < partEndOffset {
+			partWriteStartOffset = int(startOffset - partStartOffset)
+		}
+		if endOffset > partStartOffset && endOffset < partEndOffset {
+			partWriteEndOffset = int(endOffset - partStartOffset)
+		}
+		partMap[partIdx] = partWriteEndOffset - partWriteStartOffset
+	}
+	return partMap
+}
+
+func (s *FileStore) getDirtyCacheKeys() []cacheKey {
+	s.Lock.Lock()
+	defer s.Lock.Unlock()
+	var dirtyCacheKeys []cacheKey
+	for key, entry := range s.Cache {
+		if entry.File != nil {
+			dirtyCacheKeys = append(dirtyCacheKeys, key)
+		}
+	}
+	return dirtyCacheKeys
+}
+
+func (s *FileStore) setIsFlushing(flushing bool) {
+	s.Lock.Lock()
+	defer s.Lock.Unlock()
+	s.IsFlushing = flushing
+}
+
+// returns old value of IsFlushing
+func (s *FileStore) setUnlessFlushing() bool {
+	s.Lock.Lock()
+	defer s.Lock.Unlock()
+	if s.IsFlushing {
+		return true
+	}
+	s.IsFlushing = true
+	return false
+}
+
+func (s *FileStore) runFlushWithNewContext() (FlushStats, error) {
+	ctx, cancelFn := context.WithTimeout(context.Background(), DefaultFlushTime)
+	defer cancelFn()
+	return s.FlushCache(ctx)
+}
+
+func (s *FileStore) runFlusher() {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Printf("panic in filestore flusher: %v\n", r)
+			debug.PrintStack()
+		}
+	}()
+	for {
+		stats, err := s.runFlushWithNewContext()
+		if err != nil || stats.NumDirtyEntries > 0 {
+			log.Printf("filestore flush: %d/%d entries flushed, err:%v\n", stats.NumCommitted, stats.NumDirtyEntries, err)
+		}
+		if stopFlush.Load() {
+			log.Printf("filestore flusher stopping\n")
+			return
+		}
+		time.Sleep(DefaultFlushTime)
+	}
+}
+
+func minInt64(a, b int64) int64 {
+	if a < b {
+		return a
+	}
+	return b
+}
@@ -0,0 +1,347 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package filestore
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io/fs"
+	"sync"
+	"time"
+)
+
+type cacheKey struct {
+	ZoneId string
+	Name   string
+}
+
+type FileStore struct {
+	Lock       *sync.Mutex
+	Cache      map[cacheKey]*CacheEntry
+	IsFlushing bool
+}
+
+type DataCacheEntry struct {
+	PartIdx int
+	Data    []byte // capacity is always ZoneDataPartSize
+}
+
+// if File or DataEntries are not nil then they are dirty (need to be flushed to disk)
+type CacheEntry struct {
+	PinCount int // this is synchronzed with the FileStore lock (not the entry lock)
+
+	Lock        *sync.Mutex
+	ZoneId      string
+	Name        string
+	File        *WaveFile
+	DataEntries map[int]*DataCacheEntry
+	FlushErrors int
+}
+
+//lint:ignore U1000 used for testing
+func (e *CacheEntry) dump() string {
+	var buf bytes.Buffer
+	fmt.Fprintf(&buf, "CacheEntry [ZoneId: %q, Name: %q] PinCount: %d\n", e.ZoneId, e.Name, e.PinCount)
+	fmt.Fprintf(&buf, "  FileEntry: %v\n", e.File)
+	for idx, dce := range e.DataEntries {
+		fmt.Fprintf(&buf, "  DataEntry[%d]: %q\n", idx, string(dce.Data))
+	}
+	return buf.String()
+}
+
+func makeDataCacheEntry(partIdx int) *DataCacheEntry {
+	return &DataCacheEntry{
+		PartIdx: partIdx,
+		Data:    make([]byte, 0, partDataSize),
+	}
+}
+
+// will create new entries
+func (s *FileStore) getEntryAndPin(zoneId string, name string) *CacheEntry {
+	s.Lock.Lock()
+	defer s.Lock.Unlock()
+	entry := s.Cache[cacheKey{ZoneId: zoneId, Name: name}]
+	if entry == nil {
+		entry = makeCacheEntry(zoneId, name)
+		s.Cache[cacheKey{ZoneId: zoneId, Name: name}] = entry
+	}
+	entry.PinCount++
+	return entry
+}
+
+func (s *FileStore) unpinEntryAndTryDelete(zoneId string, name string) {
+	s.Lock.Lock()
+	defer s.Lock.Unlock()
+	entry := s.Cache[cacheKey{ZoneId: zoneId, Name: name}]
+	if entry == nil {
+		return
+	}
+	entry.PinCount--
+	if entry.PinCount <= 0 && entry.File == nil {
+		delete(s.Cache, cacheKey{ZoneId: zoneId, Name: name})
+	}
+}
+
+func (entry *CacheEntry) clear() {
+	entry.File = nil
+	entry.DataEntries = make(map[int]*DataCacheEntry)
+	entry.FlushErrors = 0
+}
+
+func (entry *CacheEntry) getOrCreateDataCacheEntry(partIdx int) *DataCacheEntry {
+	if entry.DataEntries[partIdx] == nil {
+		entry.DataEntries[partIdx] = makeDataCacheEntry(partIdx)
+	}
+	return entry.DataEntries[partIdx]
+}
+
+// returns err if file does not exist
+func (entry *CacheEntry) loadFileIntoCache(ctx context.Context) error {
+	if entry.File != nil {
+		return nil
+	}
+	file, err := entry.loadFileForRead(ctx)
+	if err != nil {
+		return err
+	}
+	entry.File = file
+	return nil
+}
+
+// does not populate the cache entry, returns err if file does not exist
+func (entry *CacheEntry) loadFileForRead(ctx context.Context) (*WaveFile, error) {
+	if entry.File != nil {
+		return entry.File, nil
+	}
+	file, err := dbGetZoneFile(ctx, entry.ZoneId, entry.Name)
+	if err != nil {
+		return nil, fmt.Errorf("error getting file: %w", err)
+	}
+	if file == nil {
+		return nil, fs.ErrNotExist
+	}
+	return file, nil
+}
+
+func withLock(s *FileStore, zoneId string, name string, fn func(*CacheEntry) error) error {
+	entry := s.getEntryAndPin(zoneId, name)
+	defer s.unpinEntryAndTryDelete(zoneId, name)
+	entry.Lock.Lock()
+	defer entry.Lock.Unlock()
+	return fn(entry)
+}
+
+func withLockRtn[T any](s *FileStore, zoneId string, name string, fn func(*CacheEntry) (T, error)) (T, error) {
+	var rtnVal T
+	rtnErr := withLock(s, zoneId, name, func(entry *CacheEntry) error {
+		var err error
+		rtnVal, err = fn(entry)
+		return err
+	})
+	return rtnVal, rtnErr
+}
+
+func (dce *DataCacheEntry) writeToPart(offset int64, data []byte) (int64, *DataCacheEntry) {
+	leftInPart := partDataSize - offset
+	toWrite := int64(len(data))
+	if toWrite > leftInPart {
+		toWrite = leftInPart
+	}
+	if int64(len(dce.Data)) < offset+toWrite {
+		dce.Data = dce.Data[:offset+toWrite]
+	}
+	copy(dce.Data[offset:], data[:toWrite])
+	return toWrite, dce
+}
+
+func (entry *CacheEntry) writeAt(offset int64, data []byte, replace bool) {
+	if replace {
+		entry.File.Size = 0
+	}
+	if entry.File.Opts.Circular {
+		startCirFileOffset := entry.File.Size - entry.File.Opts.MaxSize
+		if offset+int64(len(data)) <= startCirFileOffset {
+			// write is before the start of the circular file
+			return
+		}
+		if offset < startCirFileOffset {
+			// truncate data (from the front), update offset
+			truncateAmt := startCirFileOffset - offset
+			data = data[truncateAmt:]
+			offset += truncateAmt
+		}
+		if int64(len(data)) > entry.File.Opts.MaxSize {
+			// truncate data (from the front), update offset
+			truncateAmt := int64(len(data)) - entry.File.Opts.MaxSize
+			data = data[truncateAmt:]
+			offset += truncateAmt
+		}
+	}
+	endWriteOffset := offset + int64(len(data))
+	if replace {
+		entry.DataEntries = make(map[int]*DataCacheEntry)
+	}
+	for len(data) > 0 {
+		partIdx := int(offset / partDataSize)
+		if entry.File.Opts.Circular {
+			maxPart := int(entry.File.Opts.MaxSize / partDataSize)
+			partIdx = partIdx % maxPart
+		}
+		partOffset := offset % partDataSize
+		partData := entry.getOrCreateDataCacheEntry(partIdx)
+		nw, newDce := partData.writeToPart(partOffset, data)
+		entry.DataEntries[partIdx] = newDce
+		data = data[nw:]
+		offset += nw
+	}
+	if endWriteOffset > entry.File.Size || replace {
+		entry.File.Size = endWriteOffset
+	}
+	entry.File.ModTs = time.Now().UnixMilli()
+}
+
+// returns (realOffset, data, error)
+func (entry *CacheEntry) readAt(ctx context.Context, offset int64, size int64, readFull bool) (int64, []byte, error) {
+	if offset < 0 {
+		return 0, nil, fmt.Errorf("offset cannot be negative")
+	}
+	file, err := entry.loadFileForRead(ctx)
+	if err != nil {
+		return 0, nil, err
+	}
+	if readFull {
+		size = file.Size - offset
+	}
+	if offset+size > file.Size {
+		size = file.Size - offset
+	}
+	if file.Opts.Circular {
+		realDataOffset := int64(0)
+		if file.Size > file.Opts.MaxSize {
+			realDataOffset = file.Size - file.Opts.MaxSize
+		}
+		if offset < realDataOffset {
+			truncateAmt := realDataOffset - offset
+			offset += truncateAmt
+			size -= truncateAmt
+		}
+	}
+	partMap := file.computePartMap(offset, size)
+	dataEntryMap, err := entry.loadDataPartsForRead(ctx, getPartIdxsFromMap(partMap))
+	if err != nil {
+		return 0, nil, err
+	}
+	// combine the entries into a single byte slice
+	// note that we only want part of the first and last part depending on offset and size
+	rtnData := make([]byte, 0, size)
+	amtLeftToRead := size
+	curReadOffset := offset
+	for amtLeftToRead > 0 {
+		partIdx := file.partIdxAtOffset(curReadOffset)
+		partDataEntry := dataEntryMap[partIdx]
+		var partData []byte
+		if partDataEntry == nil {
+			partData = make([]byte, partDataSize)
+		} else {
+			partData = partDataEntry.Data[0:partDataSize]
+		}
+		partOffset := curReadOffset % partDataSize
+		amtToRead := minInt64(partDataSize-partOffset, amtLeftToRead)
+		rtnData = append(rtnData, partData[partOffset:partOffset+amtToRead]...)
+		amtLeftToRead -= amtToRead
+		curReadOffset += amtToRead
+	}
+	return offset, rtnData, nil
+}
+
+func prunePartsWithCache(dataEntries map[int]*DataCacheEntry, parts []int) []int {
+	var rtn []int
+	for _, partIdx := range parts {
+		if dataEntries[partIdx] != nil {
+			continue
+		}
+		rtn = append(rtn, partIdx)
+	}
+	return rtn
+}
+
+func (entry *CacheEntry) loadDataPartsIntoCache(ctx context.Context, parts []int) error {
+	parts = prunePartsWithCache(entry.DataEntries, parts)
+	if len(parts) == 0 {
+		// parts are already loaded
+		return nil
+	}
+	dbDataParts, err := dbGetFileParts(ctx, entry.ZoneId, entry.Name, parts)
+	if err != nil {
+		return fmt.Errorf("error getting data parts: %w", err)
+	}
+	for partIdx, dce := range dbDataParts {
+		entry.DataEntries[partIdx] = dce
+	}
+	return nil
+}
+
+func (entry *CacheEntry) loadDataPartsForRead(ctx context.Context, parts []int) (map[int]*DataCacheEntry, error) {
+	if len(parts) == 0 {
+		return nil, nil
+	}
+	dbParts := prunePartsWithCache(entry.DataEntries, parts)
+	var dbDataParts map[int]*DataCacheEntry
+	if len(dbParts) > 0 {
+		var err error
+		dbDataParts, err = dbGetFileParts(ctx, entry.ZoneId, entry.Name, dbParts)
+		if err != nil {
+			return nil, fmt.Errorf("error getting data parts: %w", err)
+		}
+	}
+	rtn := make(map[int]*DataCacheEntry)
+	for _, partIdx := range parts {
+		if entry.DataEntries[partIdx] != nil {
+			rtn[partIdx] = entry.DataEntries[partIdx]
+			continue
+		}
+		if dbDataParts[partIdx] != nil {
+			rtn[partIdx] = dbDataParts[partIdx]
+			continue
+		}
+		// part not found
+	}
+	return rtn, nil
+}
+
+func makeCacheEntry(zoneId string, name string) *CacheEntry {
+	return &CacheEntry{
+		Lock:        &sync.Mutex{},
+		ZoneId:      zoneId,
+		Name:        name,
+		PinCount:    0,
+		File:        nil,
+		DataEntries: make(map[int]*DataCacheEntry),
+		FlushErrors: 0,
+	}
+}
+
+func (entry *CacheEntry) flushToDB(ctx context.Context, replace bool) error {
+	if entry.File == nil {
+		return nil
+	}
+	err := dbWriteCacheEntry(ctx, entry.File, entry.DataEntries, replace)
+	if ctx.Err() != nil {
+		// transient error
+		return ctx.Err()
+	}
+	if err != nil {
+		flushErrorCount.Add(1)
+		entry.FlushErrors++
+		if entry.FlushErrors > 3 {
+			entry.clear()
+			return fmt.Errorf("too many flush errors (clearing entry): %w", err)
+		}
+		return err
+	}
+	// clear cache entry (data is now in db)
+	entry.clear()
+	return nil
+}
@@ -0,0 +1,117 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package filestore
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"github.com/wavetermdev/waveterm/wavesrv/pkg/dbutil"
+)
+
+var ErrAlreadyExists = fmt.Errorf("file already exists")
+
+func dbInsertFile(ctx context.Context, file *WaveFile) error {
+	// will fail if file already exists
+	return WithTx(ctx, func(tx *TxWrap) error {
+		query := "SELECT zoneid FROM db_wave_file WHERE zoneid = ? AND name = ?"
+		if tx.Exists(query, file.ZoneId, file.Name) {
+			return ErrAlreadyExists
+		}
+		query = "INSERT INTO db_wave_file (zoneid, name, size, createdts, modts, opts, meta) VALUES (?, ?, ?, ?, ?, ?, ?)"
+		tx.Exec(query, file.ZoneId, file.Name, file.Size, file.CreatedTs, file.ModTs, dbutil.QuickJson(file.Opts), dbutil.QuickJson(file.Meta))
+		return nil
+	})
+}
+
+func dbDeleteFile(ctx context.Context, zoneId string, name string) error {
+	return WithTx(ctx, func(tx *TxWrap) error {
+		query := "DELETE FROM db_wave_file WHERE zoneid = ? AND name = ?"
+		tx.Exec(query, zoneId, name)
+		query = "DELETE FROM db_file_data WHERE zoneid = ? AND name = ?"
+		tx.Exec(query, zoneId, name)
+		return nil
+	})
+}
+
+func dbGetZoneFileNames(ctx context.Context, zoneId string) ([]string, error) {
+	return WithTxRtn(ctx, func(tx *TxWrap) ([]string, error) {
+		var files []string
+		query := "SELECT name FROM db_wave_file WHERE zoneid = ?"
+		tx.Select(&files, query, zoneId)
+		return files, nil
+	})
+}
+
+func dbGetZoneFile(ctx context.Context, zoneId string, name string) (*WaveFile, error) {
+	return WithTxRtn(ctx, func(tx *TxWrap) (*WaveFile, error) {
+		query := "SELECT * FROM db_wave_file WHERE zoneid = ? AND name = ?"
+		file := dbutil.GetMappable[*WaveFile](tx, query, zoneId, name)
+		return file, nil
+	})
+}
+
+func dbGetAllZoneIds(ctx context.Context) ([]string, error) {
+	return WithTxRtn(ctx, func(tx *TxWrap) ([]string, error) {
+		var ids []string
+		query := "SELECT DISTINCT zoneid FROM db_wave_file"
+		tx.Select(&ids, query)
+		return ids, nil
+	})
+}
+
+func dbGetFileParts(ctx context.Context, zoneId string, name string, parts []int) (map[int]*DataCacheEntry, error) {
+	if len(parts) == 0 {
+		return nil, nil
+	}
+	return WithTxRtn(ctx, func(tx *TxWrap) (map[int]*DataCacheEntry, error) {
+		var data []*DataCacheEntry
+		query := "SELECT partidx, data FROM db_file_data WHERE zoneid = ? AND name = ? AND partidx IN (SELECT value FROM json_each(?))"
+		tx.Select(&data, query, zoneId, name, dbutil.QuickJsonArr(parts))
+		rtn := make(map[int]*DataCacheEntry)
+		for _, d := range data {
+			if cap(d.Data) != int(partDataSize) {
+				newData := make([]byte, len(d.Data), partDataSize)
+				copy(newData, d.Data)
+				d.Data = newData
+			}
+			rtn[d.PartIdx] = d
+		}
+		return rtn, nil
+	})
+}
+
+func dbGetZoneFiles(ctx context.Context, zoneId string) ([]*WaveFile, error) {
+	return WithTxRtn(ctx, func(tx *TxWrap) ([]*WaveFile, error) {
+		query := "SELECT * FROM db_wave_file WHERE zoneid = ?"
+		files := dbutil.SelectMappable[*WaveFile](tx, query, zoneId)
+		return files, nil
+	})
+}
+
+func dbWriteCacheEntry(ctx context.Context, file *WaveFile, dataEntries map[int]*DataCacheEntry, replace bool) error {
+	return WithTx(ctx, func(tx *TxWrap) error {
+		query := `SELECT zoneid FROM db_wave_file WHERE zoneid = ? AND name = ?`
+		if !tx.Exists(query, file.ZoneId, file.Name) {
+			// since deletion is synchronous this stops us from writing to a deleted file
+			return os.ErrNotExist
+		}
+		// we don't update CreatedTs or Opts
+		query = `UPDATE db_wave_file SET size = ?, modts = ?, meta = ? WHERE zoneid = ? AND name = ?`
+		tx.Exec(query, file.Size, file.ModTs, dbutil.QuickJson(file.Meta), file.ZoneId, file.Name)
+		if replace {
+			query = `DELETE FROM db_file_data WHERE zoneid = ? AND name = ?`
+			tx.Exec(query, file.ZoneId, file.Name)
+		}
+		dataPartQuery := `REPLACE INTO db_file_data (zoneid, name, partidx, data) VALUES (?, ?, ?, ?)`
+		for partIdx, dataEntry := range dataEntries {
+			if partIdx != dataEntry.PartIdx {
+				panic(fmt.Sprintf("partIdx:%d and dataEntry.PartIdx:%d do not match", partIdx, dataEntry.PartIdx))
+			}
+			tx.Exec(dataPartQuery, file.ZoneId, file.Name, dataEntry.PartIdx, dataEntry.Data)
+		}
+		return nil
+	})
+}
@@ -0,0 +1,82 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package filestore
+
+// setup for filestore db
+// includes migration support and txwrap setup
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"path"
+	"time"
+
+	"github.com/wavetermdev/thenextwave/pkg/util/migrateutil"
+	"github.com/wavetermdev/thenextwave/pkg/wavebase"
+
+	"github.com/jmoiron/sqlx"
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/sawka/txwrap"
+
+	dbfs "github.com/wavetermdev/thenextwave/db"
+)
+
+const FilestoreDBName = "filestore.db"
+
+type TxWrap = txwrap.TxWrap
+
+var globalDB *sqlx.DB
+var useTestingDb bool // just for testing (forces GetDB() to return an in-memory db)
+
+func InitFilestore() error {
+	ctx, cancelFn := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancelFn()
+	var err error
+	globalDB, err = MakeDB(ctx)
+	if err != nil {
+		return err
+	}
+	err = migrateutil.Migrate("filestore", globalDB.DB, dbfs.FilestoreMigrationFS, "migrations-filestore")
+	if err != nil {
+		return err
+	}
+	if !stopFlush.Load() {
+		go WFS.runFlusher()
+	}
+	log.Printf("filestore initialized\n")
+	return nil
+}
+
+func GetDBName() string {
+	waveHome := wavebase.GetWaveHomeDir()
+	return path.Join(waveHome, FilestoreDBName)
+}
+
+func MakeDB(ctx context.Context) (*sqlx.DB, error) {
+	var rtn *sqlx.DB
+	var err error
+	if useTestingDb {
+		dbName := ":memory:"
+		log.Printf("[db] using in-memory db\n")
+		rtn, err = sqlx.Open("sqlite3", dbName)
+	} else {
+		dbName := GetDBName()
+		log.Printf("[db] opening db %s\n", dbName)
+		rtn, err = sqlx.Open("sqlite3", fmt.Sprintf("file:%s?mode=rwc&_journal_mode=WAL&_busy_timeout=5000", dbName))
+	}
+	if err != nil {
+		return nil, fmt.Errorf("opening db: %w", err)
+	}
+	rtn.DB.SetMaxOpenConns(1)
+	return rtn, nil
+}
+
+func WithTx(ctx context.Context, fn func(tx *TxWrap) error) error {
+	return txwrap.WithTx(ctx, globalDB, fn)
+}
+
+func WithTxRtn[RT any](ctx context.Context, fn func(tx *TxWrap) (RT, error)) (RT, error) {
+	return txwrap.WithTxRtn(ctx, globalDB, fn)
+}