Compare commits

...

7 Commits

Author SHA1 Message Date
Cole Lashley 25259428ab Merge branch 'main' into cole/file-backend-changes 2024-05-06 19:44:58 -07:00
MrStashley 15f6084db4 fixed rebase artifacts 2024-05-06 19:26:42 -07:00
Ian Jennings 2772c76728 Upgrade TestDriver to v2.12.5 (#652) 2024-05-06 14:17:25 -07:00
Evan Simkowitz 4384525ca6 Remove "main" window concept from emain.ts (#651)
This continues the work to unblock multiple window support. `emain.ts`
now contains no more references to a "main" window, instead all event
handlers will dispatch to either the window that triggered the event,
the currently focused window, or the first window to be spun up.
2024-05-06 13:43:27 -07:00
Mike Sawka d6b82e3792 blockstore integration (#650)
* blockstore migration
* use a testdb for tests,
* cut down test sizes so they run a bit faster
* test init/cleanup code improvements
2024-05-06 10:24:34 -07:00
MrStashley 1dba3f2612 removed rebase artifacts 2024-05-03 18:18:01 -07:00
MrStashley 8e907066ad one commit - added fileview backend changes 2024-05-03 18:17:41 -07:00
16 changed files with 797 additions and 221 deletions
+1 -1
View File
@@ -22,7 +22,7 @@ jobs:
- uses: dashcamio/testdriver@main
id: testdriver
with:
version: v2.10.2
version: v2.12.5
prerun: |
rm ~/Desktop/WITH-LOVE-FROM-AMERICA.txt
cd ~/actions-runner/_work/testdriver/testdriver/
+31 -35
View File
@@ -34,9 +34,6 @@ let wasActive = true;
let wasInFg = true;
let currentGlobalShortcut: string | null = null;
let initialClientData: ClientDataType = null;
let windows: Windows = {};
interface Windows extends Record<string, Electron.BrowserWindow> {}
checkPromptMigrate();
ensureDir(waveHome);
@@ -325,10 +322,7 @@ function shFrameNavHandler(event: Electron.Event<Electron.WebContentsWillFrameNa
console.log("frame navigation canceled");
}
function createWindow(id: string, clientData: ClientDataType | null): Electron.BrowserWindow {
if (windows[id]) {
console.error(`createWindow called for existing window ${id}`);
}
function createWindow(clientData: ClientDataType | null): Electron.BrowserWindow {
const bounds = calcBounds(clientData);
setKeyUtilPlatform(platform());
const win = new electron.BrowserWindow({
@@ -376,18 +370,9 @@ function createWindow(id: string, clientData: ClientDataType | null): Electron.B
wasInFg = true;
wasActive = true;
});
win.on("close", () => {
delete windows[id];
});
win.webContents.on("zoom-changed", (e) => {
win.webContents.send("zoom-changed");
});
windows[id] = win;
return win;
}
function createMainWindow(clientData: ClientDataType | null) {
const win = createWindow("main", clientData);
win.webContents.setWindowOpenHandler(({ url, frameName }) => {
if (url.startsWith("https://docs.waveterm.dev/")) {
console.log("openExternal docs", url);
@@ -408,6 +393,7 @@ function createMainWindow(clientData: ClientDataType | null) {
console.log("window-open denied", url);
return { action: "deny" };
});
return win;
}
function mainResizeHandler(_: any, win: Electron.BrowserWindow) {
@@ -673,13 +659,13 @@ async function getClientData(willRetry: boolean, retryNum: number): Promise<Clie
}
function sendWSSC() {
if (windows["main"] != null) {
electron.BrowserWindow.getAllWindows().forEach((win) => {
if (waveSrvProc == null) {
windows["main"].webContents.send("wavesrv-status-change", false);
return;
win.webContents.send("wavesrv-status-change", false);
} else {
win.webContents.send("wavesrv-status-change", true, waveSrvProc.pid);
}
windows["main"].webContents.send("wavesrv-status-change", true, waveSrvProc.pid);
}
});
}
function runWaveSrv() {
@@ -747,7 +733,7 @@ electron.ipcMain.on("context-editmenu", (_, { x, y }, opts) => {
menu.popup({ x, y });
});
async function createMainWindowWrap() {
async function createWindowWrap() {
let clientData: ClientDataType | null = null;
try {
clientData = await getClientDataPoll(1);
@@ -755,9 +741,9 @@ async function createMainWindowWrap() {
} catch (e) {
console.log("error getting wavesrv clientdata", e.toString());
}
createMainWindow(clientData);
const win = createWindow(clientData);
if (clientData?.winsize.fullscreen) {
windows["main"].setFullScreen(true);
win.setFullScreen(true);
}
configureAutoUpdaterStartup(clientData);
}
@@ -776,7 +762,7 @@ function logActiveState() {
console.log("error logging active state", err);
});
// for next iteration
wasInFg = windows["main"]?.isFocused();
wasInFg = electron.BrowserWindow.getFocusedWindow()?.isFocused() ?? false;
wasActive = false;
}
@@ -802,9 +788,13 @@ function reregisterGlobalShortcut(shortcut: string) {
currentGlobalShortcut = null;
return;
}
const ok = electron.globalShortcut.register(shortcut, () => {
const ok = electron.globalShortcut.register(shortcut, async () => {
console.log("global shortcut triggered, showing window");
windows["main"]?.show();
if (electron.BrowserWindow.getAllWindows().length == 0) {
await createWindowWrap();
}
const winToShow = electron.BrowserWindow.getFocusedWindow() ?? electron.BrowserWindow.getAllWindows()[0];
winToShow?.show();
});
console.log("registered global shortcut", shortcut, ok ? "ok" : "failed");
if (!ok) {
@@ -829,9 +819,9 @@ let lastUpdateCheck: Date = null;
*/
function setAppUpdateStatus(status: string) {
appUpdateStatus = status;
if (windows["main"] != null) {
windows["main"].webContents.send("app-update-status", appUpdateStatus);
}
electron.BrowserWindow.getAllWindows().forEach((window) => {
window.webContents.send("app-update-status", appUpdateStatus);
});
}
/**
@@ -915,9 +905,14 @@ async function installAppUpdate() {
detail: "A new version has been downloaded. Restart the application to apply the updates.",
};
await electron.dialog.showMessageBox(windows["main"], dialogOpts).then(({ response }) => {
if (response === 0) autoUpdater.quitAndInstall();
});
const allWindows = electron.BrowserWindow.getAllWindows();
if (allWindows.length > 0) {
await electron.dialog
.showMessageBox(electron.BrowserWindow.getFocusedWindow() ?? allWindows[0], dialogOpts)
.then(({ response }) => {
if (response === 0) autoUpdater.quitAndInstall();
});
}
}
electron.ipcMain.on("install-app-update", () => fireAndForget(() => installAppUpdate()));
@@ -989,10 +984,11 @@ function configureAutoUpdater(enabled: boolean) {
}
setTimeout(runActiveTimer, 5000); // start active timer, wait 5s just to be safe
await app.whenReady();
await createMainWindowWrap();
await createWindowWrap();
app.on("activate", () => {
if (electron.BrowserWindow.getAllWindows().length === 0) {
createMainWindowWrap().then();
createWindowWrap().then();
}
checkForUpdates();
});
+55 -1
View File
@@ -64,6 +64,8 @@ const (
FileStatPacketStr = "filestat"
LogPacketStr = "log" // logging packet (sent from waveshell back to server)
ShellStatePacketStr = "shellstate"
ListDirPacketStr = "listdir"
SearchDirPacketStr = "searchdir"
RpcInputPacketStr = "rpcinput" // rpc-followup
SudoRequestPacketStr = "sudorequest"
SudoResponsePacketStr = "sudoresponse"
@@ -120,6 +122,8 @@ func init() {
TypeStrToFactory[WriteFileDonePacketStr] = reflect.TypeOf(WriteFileDonePacketType{})
TypeStrToFactory[LogPacketStr] = reflect.TypeOf(LogPacketType{})
TypeStrToFactory[ShellStatePacketStr] = reflect.TypeOf(ShellStatePacketType{})
TypeStrToFactory[ListDirPacketStr] = reflect.TypeOf(ListDirPacketType{})
TypeStrToFactory[SearchDirPacketStr] = reflect.TypeOf(SearchDirPacketType{})
TypeStrToFactory[FileStatPacketStr] = reflect.TypeOf(FileStatPacketType{})
TypeStrToFactory[RpcInputPacketStr] = reflect.TypeOf(RpcInputPacketType{})
TypeStrToFactory[SudoRequestPacketStr] = reflect.TypeOf(SudoRequestPacketType{})
@@ -133,6 +137,8 @@ func init() {
var _ RpcPacketType = (*ReInitPacketType)(nil)
var _ RpcPacketType = (*StreamFilePacketType)(nil)
var _ RpcPacketType = (*WriteFilePacketType)(nil)
var _ RpcPacketType = (*ListDirPacketType)(nil)
var _ RpcPacketType = (*SearchDirPacketType)(nil)
var _ RpcResponsePacketType = (*CmdStartPacketType)(nil)
var _ RpcResponsePacketType = (*ResponsePacketType)(nil)
@@ -449,8 +455,12 @@ func MakeFileStatPacketType() *FileStatPacketType {
return &FileStatPacketType{Type: FileStatPacketStr}
}
func MakeFileStatPacketFromFileInfo(finfo fs.FileInfo, err string, done bool) *FileStatPacketType {
func MakeFileStatPacketFromFileInfo(listDirPk *ListDirPacketType, finfo fs.FileInfo, err string, done bool) *FileStatPacketType {
resp := MakeFileStatPacketType()
if listDirPk != nil {
resp.RespId = listDirPk.ReqId
resp.Path = listDirPk.Path
}
resp.Error = err
resp.Done = done
@@ -464,6 +474,50 @@ func MakeFileStatPacketFromFileInfo(finfo fs.FileInfo, err string, done bool) *F
return resp
}
type ListDirPacketType struct {
Type string `json:"type"`
ReqId string `json:"reqid"`
Path string `json:"path"`
}
func (*ListDirPacketType) GetType() string {
return ListDirPacketStr
}
func (p *ListDirPacketType) GetReqId() string {
return p.ReqId
}
func MakeListDirPacket() *ListDirPacketType {
return &ListDirPacketType{Type: ListDirPacketStr}
}
type SearchDirPacketType struct {
Type string `json:"type"`
ReqId string `json:"reqid"`
Path string `json:"path"`
SearchQuery string `json:"searchquery"`
}
func (*SearchDirPacketType) GetType() string {
return SearchDirPacketStr
}
func (p *SearchDirPacketType) GetReqId() string {
return p.ReqId
}
func (p *SearchDirPacketType) ConvertToListDir() *ListDirPacketType {
rtn := MakeListDirPacket()
rtn.ReqId = p.ReqId
rtn.Path = p.Path
return rtn
}
func MakeSearchDirPacket() *SearchDirPacketType {
return &SearchDirPacketType{Type: SearchDirPacketStr}
}
type StreamFilePacketType struct {
Type string `json:"type"`
ReqId string `json:"reqid"`
+100
View File
@@ -12,6 +12,7 @@ import (
"os"
"os/exec"
"path/filepath"
"regexp"
"sort"
"strings"
"sync"
@@ -660,6 +661,97 @@ func (m *MServer) streamFile(pk *packet.StreamFilePacketType) {
return
}
func (m *MServer) writeListDirErrPacket(err error, reqId string) {
resp := packet.MakeFileStatPacketType()
resp.RespId = reqId
resp.Error = fmt.Sprintf("Error in list dir: %v", err)
resp.Done = true
m.Sender.SendPacket(resp)
}
func (m *MServer) ListDir(listDirPk *packet.ListDirPacketType) {
dirEntries, err := os.ReadDir(listDirPk.Path)
var readDirError string = ""
if err != nil {
readDirError = fmt.Sprintf("error in list dir: %v", err)
}
curDirStat, err := os.Stat(listDirPk.Path)
if err != nil {
m.writeListDirErrPacket(err, listDirPk.ReqId)
}
resp := packet.MakeFileStatPacketFromFileInfo(listDirPk, curDirStat, readDirError, false)
resp.Name = "."
m.Sender.SendPacket(resp)
curDirStat, err = os.Stat(filepath.Join(listDirPk.Path, ".."))
if err != nil {
m.writeListDirErrPacket(err, listDirPk.ReqId)
return
}
resp = packet.MakeFileStatPacketFromFileInfo(listDirPk, curDirStat, readDirError, len(dirEntries) == 0)
resp.Name = ".."
m.Sender.SendPacket(resp)
for index := 0; index < len(dirEntries); index++ {
dirEntry := dirEntries[index]
dirEntryFileInfo, err := dirEntry.Info()
if err != nil {
m.writeListDirErrPacket(err, listDirPk.ReqId)
return
}
done := index == len(dirEntries)-1
resp = packet.MakeFileStatPacketFromFileInfo(listDirPk, dirEntryFileInfo, readDirError, done)
m.Sender.SendPacket(resp)
}
}
func (m *MServer) SearchDir(searchDirPk *packet.SearchDirPacketType) {
searchEmpty := true
foundRoot := false
err := filepath.WalkDir(searchDirPk.Path, func(path string, dirEntry fs.DirEntry, err error) error {
if err != nil {
errString := fmt.Sprintf("%v", err)
if strings.Contains(errString, "operation not permitted") {
return filepath.SkipDir
}
}
fileName := filepath.Base(path)
match, err := regexp.MatchString(searchDirPk.SearchQuery, fileName)
if err != nil {
return err
}
if match {
base.Logf("matched file: %v %v", path, searchDirPk.SearchQuery)
// special case where walkdir includes the current path, which messes up the stat pk
rootName := filepath.Base(searchDirPk.Path)
if !foundRoot && fileName == rootName {
foundRoot = true
return nil
}
dirEntryFileInfo, err := dirEntry.Info()
if err != nil {
return err
}
searchEmpty = false
resp := packet.MakeFileStatPacketFromFileInfo(searchDirPk.ConvertToListDir(), dirEntryFileInfo, "", false)
m.Sender.SendPacket(resp)
}
return nil
})
if err != nil {
m.writeListDirErrPacket(err, searchDirPk.ReqId)
} else {
searchError := ""
if searchEmpty {
searchError = "none"
}
resp := packet.MakeFileStatPacketType()
resp.Error = searchError
resp.Done = true
m.Sender.SendPacket(resp)
}
}
func int64Min(v1 int64, v2 int64) int64 {
if v1 < v2 {
return v1
@@ -703,6 +795,14 @@ func (m *MServer) ProcessRpcPacket(pk packet.RpcPacketType) {
go m.writeFile(writePk, wfc)
return
}
if listDirPk, ok := pk.(*packet.ListDirPacketType); ok {
go m.ListDir(listDirPk)
return
}
if searchDirPk, ok := pk.(*packet.SearchDirPacketType); ok {
go m.SearchDir(searchDirPk)
return
}
m.Sender.SendErrorResponse(reqId, fmt.Errorf("invalid rpc type '%s'", pk.GetType()))
}
+6 -1
View File
@@ -1033,7 +1033,7 @@ func configDirHandler(w http.ResponseWriter, r *http.Request) {
var files []*packet.FileStatPacketType
for index := 0; index < len(entries); index++ {
curEntry := entries[index]
curFile := packet.MakeFileStatPacketFromFileInfo(curEntry, "", false)
curFile := packet.MakeFileStatPacketFromFileInfo(nil, curEntry, "", false)
files = append(files, curFile)
}
dirListJson, err := json.Marshal(files)
@@ -1128,6 +1128,11 @@ func main() {
log.Printf("[error] migrate up: %v\n", err)
return
}
// err = blockstore.MigrateBlockstore()
// if err != nil {
// log.Printf("[error] migrate blockstore: %v\n", err)
// return
// }
clientData, err := sstore.EnsureClientData(context.Background())
if err != nil {
log.Printf("[error] ensuring client data: %v\n", err)
@@ -0,0 +1 @@
-- nothing
@@ -0,0 +1,19 @@
CREATE TABLE block_file (
blockid varchar(36) NOT NULL,
name varchar(200) NOT NULL,
maxsize bigint NOT NULL,
circular boolean NOT NULL,
size bigint NOT NULL,
createdts bigint NOT NULL,
modts bigint NOT NULL,
meta json NOT NULL,
PRIMARY KEY (blockid, name)
);
CREATE TABLE block_data (
blockid varchar(36) NOT NULL,
name varchar(200) NOT NULL,
partidx int NOT NULL,
data blob NOT NULL,
PRIMARY KEY(blockid, name, partidx)
);
+3
View File
@@ -10,3 +10,6 @@ import "embed"
//go:embed migrations/*.sql
var MigrationFS embed.FS
//go:embed blockstore-migrations/*.sql
var BlockstoreMigrationFS embed.FS
-1
View File
@@ -5,7 +5,6 @@ go 1.22
toolchain go1.22.0
require (
github.com/alecthomas/units v0.0.0-20231202071711-9a357b53e9c9
github.com/alessio/shellescape v1.4.1
github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2
github.com/creack/pty v1.1.18
-5
View File
@@ -1,5 +1,3 @@
github.com/alecthomas/units v0.0.0-20231202071711-9a357b53e9c9 h1:ez/4by2iGztzR4L0zgAOR8lTQK9VlyBVVd7G4omaOQs=
github.com/alecthomas/units v0.0.0-20231202071711-9a357b53e9c9/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
github.com/alessio/shellescape v1.4.1 h1:V7yhSDDn8LP4lc4jS8pFkt0zCnzVJlG5JXy9BVKJUX0=
github.com/alessio/shellescape v1.4.1/go.mod h1:PZAiSCk0LJaZkiCSkPv8qIobYglO3FPpyFjDCtHLS30=
github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2 h1:7Ip0wMmLHLRJdrloDxZfhMm0xrLXZS8+COSu2bXmEQs=
@@ -57,7 +55,6 @@ github.com/sawka/txwrap v0.1.2 h1:v8xS0Z1LE7/6vMZA81PYihI+0TSR6Zm1MalzzBIuXKc=
github.com/sawka/txwrap v0.1.2/go.mod h1:T3nlw2gVpuolo6/XEetvBbk1oMXnY978YmBFy1UyHvw=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/wavetermdev/ssh_config v0.0.0-20240306041034-17e2087ebde2 h1:onqZrJVap1sm15AiIGTfWzdr6cEF0KdtddeuuOVhzyY=
@@ -74,8 +71,6 @@ golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.15.0 h1:y/Oo/a/q3IXu26lQgl04j/gjuBDOBlx7X6Om1j2CPW4=
golang.org/x/term v0.15.0/go.mod h1:BDl952bC7+uMoWR75FIrCDx79TPU9oHkTZ9yRbYOrX0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
mvdan.cc/sh/v3 v3.7.0 h1:lSTjdP/1xsddtaKfGg7Myu7DnlHItd3/M2tomOcNNBg=
+37 -28
View File
@@ -10,8 +10,6 @@ import (
"strings"
"sync"
"time"
"github.com/alecthomas/units"
)
type FileOptsType struct {
@@ -32,7 +30,11 @@ type FileInfo struct {
Meta FileMeta
}
const MaxBlockSize = int64(128 * units.Kilobyte)
const UnitsKB = 1024 * 1024
const UnitsMB = 1024 * UnitsKB
const UnitsGB = 1024 * UnitsMB
const MaxBlockSize = int64(128 * UnitsKB)
const DefaultFlushTimeout = 1 * time.Second
type CacheEntry struct {
@@ -79,16 +81,23 @@ type BlockStore interface {
GetAllBlockIds(ctx context.Context) []string
}
var cache map[string]*CacheEntry = make(map[string]*CacheEntry)
var blockstoreCache map[string]*CacheEntry = make(map[string]*CacheEntry)
var globalLock *sync.Mutex = &sync.Mutex{}
var appendLock *sync.Mutex = &sync.Mutex{}
var flushTimeout = DefaultFlushTimeout
var lastWriteTime time.Time
// for testing
func clearCache() {
globalLock.Lock()
defer globalLock.Unlock()
blockstoreCache = make(map[string]*CacheEntry)
}
func InsertFileIntoDB(ctx context.Context, fileInfo FileInfo) error {
metaJson, err := json.Marshal(fileInfo.Meta)
if err != nil {
return fmt.Errorf("Error writing file %s to db: %v", fileInfo.Name, err)
return fmt.Errorf("error writing file %s to db: %v", fileInfo.Name, err)
}
txErr := WithTx(ctx, func(tx *TxWrap) error {
query := `INSERT INTO block_file VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
@@ -96,7 +105,7 @@ func InsertFileIntoDB(ctx context.Context, fileInfo FileInfo) error {
return nil
})
if txErr != nil {
return fmt.Errorf("Error writing file %s to db: %v", fileInfo.Name, txErr)
return fmt.Errorf("error writing file %s to db: %v", fileInfo.Name, txErr)
}
return nil
}
@@ -104,7 +113,7 @@ func InsertFileIntoDB(ctx context.Context, fileInfo FileInfo) error {
func WriteFileToDB(ctx context.Context, fileInfo FileInfo) error {
metaJson, err := json.Marshal(fileInfo.Meta)
if err != nil {
return fmt.Errorf("Error writing file %s to db: %v", fileInfo.Name, err)
return fmt.Errorf("error writing file %s to db: %v", fileInfo.Name, err)
}
txErr := WithTx(ctx, func(tx *TxWrap) error {
query := `UPDATE block_file SET blockid = ?, name = ?, maxsize = ?, circular = ?, size = ?, createdts = ?, modts = ?, meta = ? where blockid = ? and name = ?`
@@ -112,7 +121,7 @@ func WriteFileToDB(ctx context.Context, fileInfo FileInfo) error {
return nil
})
if txErr != nil {
return fmt.Errorf("Error writing file %s to db: %v", fileInfo.Name, txErr)
return fmt.Errorf("error writing file %s to db: %v", fileInfo.Name, txErr)
}
return nil
@@ -125,7 +134,7 @@ func WriteDataBlockToDB(ctx context.Context, blockId string, name string, index
return nil
})
if txErr != nil {
return fmt.Errorf("Error writing data block to db: %v", txErr)
return fmt.Errorf("error writing data block to db: %v", txErr)
}
return nil
}
@@ -152,7 +161,7 @@ func WriteToCacheBlockNum(ctx context.Context, blockId string, name string, p []
defer cacheEntry.Lock.Unlock()
block, err := GetCacheBlock(ctx, blockId, name, cacheNum, pullFromDB)
if err != nil {
return 0, 0, fmt.Errorf("Error getting cache block: %v", err)
return 0, 0, fmt.Errorf("error getting cache block: %v", err)
}
var bytesWritten = 0
blockLen := len(block.data)
@@ -192,7 +201,7 @@ func ReadFromCacheBlock(ctx context.Context, blockId string, name string, block
}
}()
if pos > len(block.data) {
return 0, fmt.Errorf("Reading past end of cache block, should never happen")
return 0, fmt.Errorf("reading past end of cache block, should never happen")
}
bytesWritten := 0
index := pos
@@ -216,7 +225,7 @@ func ReadFromCacheBlock(ctx context.Context, blockId string, name string, block
return bytesWritten, nil
}
const MaxSizeError = "Hit Max Size"
const MaxSizeError = "MaxSizeError"
func WriteToCacheBuf(buf *[]byte, p []byte, pos int, length int, maxWrite int64) (int, error) {
bytesToWrite := length
@@ -260,7 +269,7 @@ func GetValuesFromCacheId(cacheId string) (blockId string, name string) {
func GetCacheEntry(ctx context.Context, blockId string, name string) (*CacheEntry, bool) {
globalLock.Lock()
defer globalLock.Unlock()
if curCacheEntry, found := cache[GetCacheId(blockId, name)]; found {
if curCacheEntry, found := blockstoreCache[GetCacheId(blockId, name)]; found {
return curCacheEntry, true
} else {
return nil, false
@@ -279,7 +288,7 @@ func GetCacheEntryOrPopulate(ctx context.Context, blockId string, name string) (
if cacheEntry, found := GetCacheEntry(ctx, blockId, name); found {
return cacheEntry, nil
} else {
return nil, fmt.Errorf("Error getting cache entry %v %v", blockId, name)
return nil, fmt.Errorf("error getting cache entry %v %v", blockId, name)
}
}
@@ -288,16 +297,16 @@ func GetCacheEntryOrPopulate(ctx context.Context, blockId string, name string) (
func SetCacheEntry(ctx context.Context, cacheId string, cacheEntry *CacheEntry) {
globalLock.Lock()
defer globalLock.Unlock()
if _, found := cache[cacheId]; found {
if _, found := blockstoreCache[cacheId]; found {
return
}
cache[cacheId] = cacheEntry
blockstoreCache[cacheId] = cacheEntry
}
func DeleteCacheEntry(ctx context.Context, blockId string, name string) {
globalLock.Lock()
defer globalLock.Unlock()
delete(cache, GetCacheId(blockId, name))
delete(blockstoreCache, GetCacheId(blockId, name))
}
func GetCacheBlock(ctx context.Context, blockId string, name string, cacheNum int, pullFromDB bool) (*CacheBlock, error) {
@@ -392,7 +401,7 @@ func WriteAtHelper(ctx context.Context, blockId string, name string, p []byte, o
}
fInfo, err := Stat(ctx, blockId, name)
if err != nil {
return 0, fmt.Errorf("Write At err: %v", err)
return 0, fmt.Errorf("WriteAt err: %v", err)
}
if off > fInfo.Opts.MaxSize && fInfo.Opts.Circular {
numOver := off / fInfo.Opts.MaxSize
@@ -416,12 +425,12 @@ func WriteAtHelper(ctx context.Context, blockId string, name string, p []byte, o
b, err := WriteAtHelper(ctx, blockId, name, p, 0, false)
bytesWritten += b
if err != nil {
return bytesWritten, fmt.Errorf("Write to cache error: %v", err)
return bytesWritten, fmt.Errorf("write to cache error: %v", err)
}
break
}
} else {
return bytesWritten, fmt.Errorf("Write to cache error: %v", err)
return bytesWritten, fmt.Errorf("write to cache error: %v", err)
}
}
if len(p) == b {
@@ -452,7 +461,7 @@ func GetAllBlockSizes(dataBlocks []*CacheBlock) (int, int) {
}
func FlushCache(ctx context.Context) error {
for _, cacheEntry := range cache {
for _, cacheEntry := range blockstoreCache {
err := WriteFileToDB(ctx, *cacheEntry.Info)
if err != nil {
return err
@@ -485,14 +494,14 @@ func ReadAt(ctx context.Context, blockId string, name string, p *[]byte, off int
bytesRead := 0
fInfo, err := Stat(ctx, blockId, name)
if err != nil {
return 0, fmt.Errorf("Read At err: %v", err)
return 0, fmt.Errorf("ReadAt err: %v", err)
}
if off > fInfo.Opts.MaxSize && fInfo.Opts.Circular {
numOver := off / fInfo.Opts.MaxSize
off = off - (numOver * fInfo.Opts.MaxSize)
}
if off > fInfo.Size {
return 0, fmt.Errorf("Read At error: tried to read past the end of the file")
return 0, fmt.Errorf("ReadAt error: tried to read past the end of the file")
}
endReadPos := math.Min(float64(int64(len(*p))+off), float64(fInfo.Size))
bytesToRead := int64(endReadPos) - off
@@ -505,7 +514,7 @@ func ReadAt(ctx context.Context, blockId string, name string, p *[]byte, off int
for index := curCacheNum; index < curCacheNum+numCaches; index++ {
curCacheBlock, err := GetCacheBlock(ctx, blockId, name, index, true)
if err != nil {
return bytesRead, fmt.Errorf("Error getting cache block: %v", err)
return bytesRead, fmt.Errorf("error getting cache block: %v", err)
}
cacheOffset := off - (int64(index) * MaxBlockSize)
if cacheOffset < 0 {
@@ -540,7 +549,7 @@ func ReadAt(ctx context.Context, blockId string, name string, p *[]byte, off int
break
}
} else {
return bytesRead, fmt.Errorf("Read from cache error: %v", err)
return bytesRead, fmt.Errorf("read from cache error: %v", err)
}
}
}
@@ -552,7 +561,7 @@ func AppendData(ctx context.Context, blockId string, name string, p []byte) (int
defer appendLock.Unlock()
fInfo, err := Stat(ctx, blockId, name)
if err != nil {
return 0, fmt.Errorf("Append stat error: %v", err)
return 0, fmt.Errorf("append stat error: %v", err)
}
return WriteAt(ctx, blockId, name, p, fInfo.Size)
}
@@ -564,12 +573,12 @@ func DeleteFile(ctx context.Context, blockId string, name string) error {
}
func DeleteBlock(ctx context.Context, blockId string) error {
for cacheId, _ := range cache {
for cacheId := range blockstoreCache {
curBlockId, name := GetValuesFromCacheId(cacheId)
if curBlockId == blockId {
err := DeleteFile(ctx, blockId, name)
if err != nil {
return fmt.Errorf("Error deleting %v %v: %v", blockId, name, err)
return fmt.Errorf("error deleting %v %v: %v", blockId, name, err)
}
}
}
+64 -3
View File
@@ -8,11 +8,16 @@ import (
"path"
"sync"
"github.com/golang-migrate/migrate/v4"
_ "github.com/golang-migrate/migrate/v4/database/sqlite3"
"github.com/golang-migrate/migrate/v4/source/iofs"
"github.com/jmoiron/sqlx"
_ "github.com/mattn/go-sqlite3"
"github.com/sawka/txwrap"
"github.com/wavetermdev/waveterm/wavesrv/pkg/dbutil"
"github.com/wavetermdev/waveterm/wavesrv/pkg/scbase"
dbfs "github.com/wavetermdev/waveterm/wavesrv/db"
)
const DBFileName = "blockstore.db"
@@ -21,12 +26,64 @@ type SingleConnDBGetter struct {
SingleConnLock *sync.Mutex
}
var dbWrap *SingleConnDBGetter
var dbWrap *SingleConnDBGetter = &SingleConnDBGetter{SingleConnLock: &sync.Mutex{}}
type TxWrap = txwrap.TxWrap
func InitDBState() {
dbWrap = &SingleConnDBGetter{SingleConnLock: &sync.Mutex{}}
func MakeBlockstoreMigrate() (*migrate.Migrate, error) {
fsVar, err := iofs.New(dbfs.BlockstoreMigrationFS, "blockstore-migrations")
if err != nil {
return nil, fmt.Errorf("opening iofs: %w", err)
}
dbUrl := fmt.Sprintf("sqlite3://%s", GetDBName())
m, err := migrate.NewWithSourceInstance("iofs", fsVar, dbUrl)
if err != nil {
return nil, fmt.Errorf("making blockstore migration db[%s]: %w", GetDBName(), err)
}
return m, nil
}
func MigrateBlockstore() error {
log.Printf("migrate blockstore\n")
m, err := MakeBlockstoreMigrate()
if err != nil {
return err
}
curVersion, dirty, err := GetMigrateVersion(m)
if dirty {
return fmt.Errorf("cannot migrate up, database is dirty")
}
if err != nil {
return fmt.Errorf("cannot get current migration version: %v", err)
}
defer m.Close()
err = m.Up()
if err != nil && err != migrate.ErrNoChange {
return fmt.Errorf("migrating blockstore: %w", err)
}
newVersion, _, err := GetMigrateVersion(m)
if err != nil {
return fmt.Errorf("cannot get new migration version: %v", err)
}
if newVersion != curVersion {
log.Printf("[db] blockstore migration done, version %d -> %d\n", curVersion, newVersion)
}
return nil
}
func GetMigrateVersion(m *migrate.Migrate) (uint, bool, error) {
if m == nil {
var err error
m, err = MakeBlockstoreMigrate()
if err != nil {
return 0, false, err
}
}
curVersion, dirty, err := m.Version()
if err == migrate.ErrNilVersion {
return 0, false, nil
}
return curVersion, dirty, err
}
func (dbg *SingleConnDBGetter) GetDB(ctx context.Context) (*sqlx.DB, error) {
@@ -62,8 +119,12 @@ func WithTxRtn[RT any](ctx context.Context, fn func(tx *TxWrap) (RT, error)) (RT
var globalDBLock = &sync.Mutex{}
var globalDB *sqlx.DB
var globalDBErr error
var overrideDBName string
func GetDBName() string {
if overrideDBName != "" {
return overrideDBName
}
scHome := scbase.GetWaveHomeDir()
return path.Join(scHome, DBFileName)
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+9
View File
@@ -1522,6 +1522,14 @@ func (wsh *WaveshellProc) StreamFile(ctx context.Context, streamPk *packet.Strea
return wsh.PacketRpcIter(ctx, streamPk)
}
func (msh *WaveshellProc) ListDir(ctx context.Context, listDirPk *packet.ListDirPacketType) (*packet.RpcResponseIter, error) {
return msh.PacketRpcIter(ctx, listDirPk)
}
func (msh *WaveshellProc) SearchDir(ctx context.Context, searchDirPk *packet.SearchDirPacketType) (*packet.RpcResponseIter, error) {
return msh.PacketRpcIter(ctx, searchDirPk)
}
func addScVarsToState(state *packet.ShellState) *packet.ShellState {
if state == nil {
return nil
@@ -2210,6 +2218,7 @@ func (wsh *WaveshellProc) PacketRpcIter(ctx context.Context, pk packet.RpcPacket
if pk == nil {
return nil, fmt.Errorf("PacketRpc passed nil packet")
}
log.Printf("sending packet: %v", pk)
reqId := pk.GetReqId()
wsh.ServerProc.Output.RegisterRpcSz(reqId, RpcIterChannelSize)
err := wsh.ServerProc.Input.SendPacketCtx(ctx, pk)
+2 -2
View File
@@ -17,7 +17,7 @@ import (
_ "github.com/golang-migrate/migrate/v4/source/file"
"github.com/golang-migrate/migrate/v4/source/iofs"
_ "github.com/mattn/go-sqlite3"
sh2db "github.com/wavetermdev/waveterm/wavesrv/db"
dbfs "github.com/wavetermdev/waveterm/wavesrv/db"
"github.com/golang-migrate/migrate/v4"
)
@@ -29,7 +29,7 @@ const CmdLineSpecialMigration = 20
const RISpecialMigration = 30
func MakeMigrate() (*migrate.Migrate, error) {
fsVar, err := iofs.New(sh2db.MigrationFS, "migrations")
fsVar, err := iofs.New(dbfs.MigrationFS, "migrations")
if err != nil {
return nil, fmt.Errorf("opening iofs: %w", err)
}