waveterm/backup
0
0
Fork 0
mirror of https://github.com/wavetermdev/backup.git synced 2026-08-05 13:57:07 -07:00
Code Issues Packages Projects Releases Wiki Activity

zsh reinit fixes (#477)

Browse Source 
* reset command now initiates and completes async so there is feedback that something is happening when it takes a long time

* switch from standard rpc to rpciter

* checkpoint on reinit -- stream output, stats packet, logging to cmd pty, new endBytes for EOF

* make generic versions of endbytes scanner and channel output funcs

* update bash to use more modern state parsing (tricks learned from zsh)

* verbose mode, fix stats output message

* add a diff when verbose mode is on
This commit is contained in:
Mike Sawka
2024-03-19 16:38:38 -07:00
committed by GitHub
parent accb74ae0f
commit 5616c9abbb
11 changed files with 427 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wavesrv/pkg/cmdrunner/cmdrunner.go
+69 -18
View File
@@ -41,6 +41,7 @@ import (
"github.com/wavetermdev/waveterm/wavesrv/pkg/releasechecker"
"github.com/wavetermdev/waveterm/wavesrv/pkg/remote"
"github.com/wavetermdev/waveterm/wavesrv/pkg/remote/openai"
"github.com/wavetermdev/waveterm/wavesrv/pkg/rtnstate"
"github.com/wavetermdev/waveterm/wavesrv/pkg/scbase"
"github.com/wavetermdev/waveterm/wavesrv/pkg/scbus"
"github.com/wavetermdev/waveterm/wavesrv/pkg/scpacket"
@@ -1648,8 +1649,12 @@ func CopyFileCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scb
}
var outputPos int64
outputStr := fmt.Sprintf("Copying [%v]:%v to [%v]:%v\r\n", sourceRemoteId.DisplayName, sourceFullPath, destRemoteId.DisplayName, destFullPath)
termopts := sstore.TermOpts{Rows: shellutil.DefaultTermRows, Cols: shellutil.DefaultTermCols, FlexRows: true, MaxPtySize: remote.DefaultMaxPtySize}
cmd, err := makeDynCmd(ctx, "copy file", ids, pk.GetRawStr(), termopts)
termOpts, err := GetUITermOpts(pk.UIContext.WinSize, DefaultPTERM)
if err != nil {
return nil, fmt.Errorf("cannot make termopts: %w", err)
}
pkTermOpts := convertTermOpts(termOpts)
cmd, err := makeDynCmd(ctx, "copy file", ids, pk.GetRawStr(), *pkTermOpts)
writeStringToPty(ctx, cmd, outputStr, &outputPos)
if err != nil {
// TODO tricky error since the command was a success, but we can't show the output
@@ -3655,11 +3660,14 @@ func SessionCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbu
return update, nil
}
func RemoteResetCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.UpdatePacket, error) {
func RemoteResetCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (rtnUpdate scbus.UpdatePacket, rtnErr error) {
ids, err := resolveUiIds(ctx, pk, R_Session|R_Screen|R_Remote)
if err != nil {
return nil, err
}
if !ids.Remote.MShell.IsConnected() {
return nil, fmt.Errorf("cannot reinit, remote is not connected")
}
shellType := ids.Remote.ShellType
if pk.Kwargs["shell"] != "" {
shellArg := pk.Kwargs["shell"]
@@ -3668,33 +3676,76 @@ func RemoteResetCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (
}
shellType = shellArg
}
ssPk, err := ids.Remote.MShell.ReInit(ctx, shellType)
verbose := resolveBool(pk.Kwargs["verbose"], false)
termOpts, err := GetUITermOpts(pk.UIContext.WinSize, DefaultPTERM)
if err != nil {
return nil, err
return nil, fmt.Errorf("cannot make termopts: %w", err)
}
if ssPk == nil || ssPk.State == nil {
return nil, fmt.Errorf("invalid initpk received from remote (no remote state)")
}
feState := sstore.FeStateFromShellState(ssPk.State)
remoteInst, err := sstore.UpdateRemoteState(ctx, ids.SessionId, ids.ScreenId, ids.Remote.RemotePtr, feState, ssPk.State, nil)
pkTermOpts := convertTermOpts(termOpts)
cmd, err := makeDynCmd(ctx, "reset", ids, pk.GetRawStr(), *pkTermOpts)
if err != nil {
return nil, err
}
outputStr := fmt.Sprintf("reset remote state (shell:%s)", ssPk.State.GetShellType())
cmd, err := makeStaticCmd(ctx, "reset", ids, pk.GetRawStr(), []byte(outputStr))
if err != nil {
// TODO tricky error since the command was a success, but we can't show the output
return nil, err
}
update, err := addLineForCmd(ctx, "/reset", false, ids, cmd, "", nil)
if err != nil {
// TODO tricky error since the command was a success, but we can't show the output
return nil, err
}
update.AddUpdate(sstore.MakeSessionUpdateForRemote(ids.SessionId, remoteInst), sstore.InteractiveUpdate(pk.Interactive))
go doResetCommand(ids, shellType, cmd, verbose)
return update, nil
}
func doResetCommand(ids resolvedIds, shellType string, cmd *sstore.CmdType, verbose bool) {
ctx, cancelFn := context.WithTimeout(context.Background(), 10*time.Second)
defer cancelFn()
startTime := time.Now()
var outputPos int64
var rtnErr error
exitSuccess := true
defer func() {
if rtnErr != nil {
exitSuccess = false
writeStringToPty(ctx, cmd, fmt.Sprintf("\r\nerror: %v", rtnErr), &outputPos)
}
deferWriteCmdStatus(ctx, cmd, startTime, exitSuccess, outputPos)
}()
dataFn := func(data []byte) {
writeStringToPty(ctx, cmd, string(data), &outputPos)
}
origStatePtr := ids.Remote.MShell.GetDefaultStatePtr(shellType)
ssPk, err := ids.Remote.MShell.ReInit(ctx, shellType, dataFn, verbose)
if err != nil {
rtnErr = err
return
}
if ssPk == nil || ssPk.State == nil {
rtnErr = fmt.Errorf("invalid initpk received from remote (no remote state)")
return
}
feState := sstore.FeStateFromShellState(ssPk.State)
remoteInst, err := sstore.UpdateRemoteState(ctx, ids.SessionId, ids.ScreenId, ids.Remote.RemotePtr, feState, ssPk.State, nil)
if err != nil {
rtnErr = err
return
}
newStatePtr := ids.Remote.MShell.GetDefaultStatePtr(shellType)
if verbose && origStatePtr != nil && newStatePtr != nil {
statePtrDiff := fmt.Sprintf("oldstate: %v, newstate: %v\r\n", origStatePtr.BaseHash, newStatePtr.BaseHash)
writeStringToPty(ctx, cmd, statePtrDiff, &outputPos)
origFullState, _ := sstore.GetFullState(ctx, *origStatePtr)
newFullState, _ := sstore.GetFullState(ctx, *newStatePtr)
if origFullState != nil && newFullState != nil {
var diffBuf bytes.Buffer
rtnstate.DisplayStateUpdateDiff(&diffBuf, *origFullState, *newFullState)
diffStr := diffBuf.String()
diffStr = strings.ReplaceAll(diffStr, "\n", "\r\n")
writeStringToPty(ctx, cmd, diffStr, &outputPos)
}
}
update := scbus.MakeUpdatePacket()
update.AddUpdate(sstore.MakeSessionUpdateForRemote(ids.SessionId, remoteInst))
scbus.MainUpdateBus.DoUpdate(update)
}
func ResetCwdCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.UpdatePacket, error) {
ids, err := resolveUiIds(ctx, pk, R_Session|R_Screen|R_Remote)
if err != nil {
wavesrv/pkg/remote/remote.go
+62 -16
View File
@@ -196,7 +196,7 @@ func (msh *MShellProc) EnsureShellType(ctx context.Context, shellType string) er
return nil
}
// try to reinit the shell
_, err := msh.ReInit(ctx, shellType)
_, err := msh.ReInit(ctx, shellType, nil, false)
if err != nil {
return fmt.Errorf("error trying to initialize shell %q: %v", shellType, err)
}
@@ -1401,33 +1401,60 @@ func makeReinitErrorUpdate(shellType string) sstore.ActivityUpdate {
return rtn
}
func (msh *MShellProc) ReInit(ctx context.Context, shellType string) (*packet.ShellStatePacketType, error) {
func (msh *MShellProc) ReInit(ctx context.Context, shellType string, dataFn func([]byte), verbose bool) (rtnPk *packet.ShellStatePacketType, rtnErr error) {
if !msh.IsConnected() {
return nil, fmt.Errorf("cannot reinit, remote is not connected")
}
if shellType != packet.ShellType_bash && shellType != packet.ShellType_zsh {
return nil, fmt.Errorf("invalid shell type %q", shellType)
}
if dataFn == nil {
dataFn = func([]byte) {}
}
defer func() {
if rtnErr != nil {
sstore.UpdateActivityWrap(ctx, makeReinitErrorUpdate(shellType), "reiniterror")
}
}()
startTs := time.Now()
reinitPk := packet.MakeReInitPacket()
reinitPk.ReqId = uuid.New().String()
reinitPk.ShellType = shellType
resp, err := msh.PacketRpcRaw(ctx, reinitPk)
rpcIter, err := msh.PacketRpcIter(ctx, reinitPk)
if err != nil {
return nil, err
}
if resp == nil {
return nil, fmt.Errorf("no response")
}
ssPk, ok := resp.(*packet.ShellStatePacketType)
if !ok {
sstore.UpdateActivityWrap(ctx, makeReinitErrorUpdate(shellType), "reiniterror")
if respPk, ok := resp.(*packet.ResponsePacketType); ok && respPk.Error != "" {
return nil, fmt.Errorf("error reinitializing remote: %s", respPk.Error)
defer rpcIter.Close()
var ssPk *packet.ShellStatePacketType
for {
resp, err := rpcIter.Next(ctx)
if err != nil {
return nil, err
}
return nil, fmt.Errorf("invalid reinit response (not an shellstate packet): %T", resp)
if resp == nil {
return nil, fmt.Errorf("channel closed with no response")
}
var ok bool
ssPk, ok = resp.(*packet.ShellStatePacketType)
if ok {
break
}
respPk, ok := resp.(*packet.ResponsePacketType)
if ok {
if respPk.Error != "" {
return nil, fmt.Errorf("error reinitializing remote: %s", respPk.Error)
}
return nil, fmt.Errorf("invalid response from waveshell")
}
dataPk, ok := resp.(*packet.FileDataPacketType)
if ok {
dataFn(dataPk.Data)
continue
}
invalidPkStr := fmt.Sprintf("\r\ninvalid packettype from waveshell: %s\r\n", resp.GetType())
dataFn([]byte(invalidPkStr))
}
if ssPk.State == nil {
sstore.UpdateActivityWrap(ctx, makeReinitErrorUpdate(shellType), "reiniterror")
if ssPk == nil || ssPk.State == nil {
return nil, fmt.Errorf("invalid reinit response shellstate packet does not contain remote state")
}
// TODO: maybe we don't need to save statebase here. should be possible to save it on demand
@@ -1438,10 +1465,29 @@ func (msh *MShellProc) ReInit(ctx context.Context, shellType string) (*packet.Sh
return nil, fmt.Errorf("error storing remote state: %w", err)
}
msh.StateMap.SetCurrentState(ssPk.State.GetShellType(), ssPk.State)
msh.WriteToPtyBuffer("initialized shell:%s state:%s\n", shellType, ssPk.State.GetHashVal(false))
timeDur := time.Since(startTs)
dataFn([]byte(makeShellInitOutputMsg(verbose, ssPk.State, ssPk.Stats, timeDur, false)))
msh.WriteToPtyBuffer("%s", makeShellInitOutputMsg(false, ssPk.State, ssPk.Stats, timeDur, true))
return ssPk, nil
}
func makeShellInitOutputMsg(verbose bool, state *packet.ShellState, stats *packet.ShellStateStats, dur time.Duration, ptyMsg bool) string {
if !verbose || ptyMsg {
if ptyMsg {
return fmt.Sprintf("initialized state shell:%s statehash:%s %dms\n", state.GetShellType(), state.GetHashVal(false), dur.Milliseconds())
} else {
return fmt.Sprintf("initialized connection state (shell:%s)\r\n", state.GetShellType())
}
}
var buf bytes.Buffer
buf.WriteString("-----\r\n")
buf.WriteString(fmt.Sprintf("initialized connection shell:%s statehash:%s %dms\r\n", state.GetShellType(), state.GetHashVal(false), dur.Milliseconds()))
if stats != nil {
buf.WriteString(fmt.Sprintf(" outsize:%s size:%s env:%d, vars:%d, aliases:%d, funcs:%d\r\n", scbase.NumFormatDec(stats.OutputSize), scbase.NumFormatDec(stats.StateSize), stats.EnvCount, stats.VarCount, stats.AliasCount, stats.FuncCount))
}
return buf.String()
}
func (msh *MShellProc) WriteFile(ctx context.Context, writePk *packet.WriteFilePacketType) (*packet.RpcResponseIter, error) {
return msh.PacketRpcIter(ctx, writePk)
}
@@ -1690,7 +1736,7 @@ func (msh *MShellProc) initActiveShells() {
return
}
for _, shellType := range activeShells {
_, err = msh.ReInit(ctx, shellType)
_, err = msh.ReInit(ctx, shellType, nil, false)
if err != nil {
msh.WriteToPtyBuffer("*error reiniting shell %q: %v\n", shellType, err)
}