From e239a0948b8afd94f440930e94ea5513782ba2e5 Mon Sep 17 00:00:00 2001 From: lyx-tec Date: Sat, 27 Jun 2026 23:58:18 +0800 Subject: [PATCH] fix: canonical term size model for shell resize Replace fire-and-forget resize events with a canonical term size state model. Term size is now block-level canonical state (stored in DB), not an event racing through the controller input channel. Key changes: - Add ApplyTermSize to Controller interface (Shell/Session/Tsunami) - Add SetTerminalSize(blockId, size): write DB canonical + best-effort apply - Add getLatestTermSize(blockId): fresh DB read for shell startup - New SetTerminalSizeCommand RPC; frontend syncControllerTermSize uses it - run() and startNewJob() read canonical size instead of rtOpts snapshot - ResyncController writes rtOpts.TermSize to canonical on entry - ControllerInputCommand handler strips TermSize (routes to SetTerminalSize) - Remove updateTermSize and term-size handling from ShellController input goroutine Invariants: - resize always writes canonical state first - shell startup always reads fresh canonical from DB --- frontend/app/store/wshclientapi.ts | 6 +++ frontend/app/view/term/termwrap.ts | 2 +- frontend/types/gotypes.d.ts | 6 +++ pkg/blockcontroller/blockcontroller.go | 52 +++++++++++++++++++ .../sessiondaemoncontroller.go | 16 +++--- pkg/blockcontroller/shellcontroller.go | 52 ++++--------------- pkg/blockcontroller/tsunamicontroller.go | 4 ++ pkg/wshrpc/wshclient/wshclient.go | 6 +++ pkg/wshrpc/wshrpctypes.go | 6 +++ pkg/wshrpc/wshserver/wshserver.go | 14 ++++- 10 files changed, 113 insertions(+), 51 deletions(-) diff --git a/frontend/app/store/wshclientapi.ts b/frontend/app/store/wshclientapi.ts index 5155a22652..956345c90c 100644 --- a/frontend/app/store/wshclientapi.ts +++ b/frontend/app/store/wshclientapi.ts @@ -924,6 +924,12 @@ export class RpcApiType { return client.wshRpcCall("setsecrets", data, opts); } + // command "setterminalsize" [call] + SetTerminalSizeCommand(client: WshClient, data: CommandSetTerminalSizeData, opts?: RpcOpts): Promise { + if (this.mockClient) return this.mockClient.mockWshRpcCall(client, "setterminalsize", data, opts); + return client.wshRpcCall("setterminalsize", data, opts); + } + // command "setvar" [call] SetVarCommand(client: WshClient, data: CommandVarData, opts?: RpcOpts): Promise { if (this.mockClient) return this.mockClient.mockWshRpcCall(client, "setvar", data, opts); diff --git a/frontend/app/view/term/termwrap.ts b/frontend/app/view/term/termwrap.ts index 814a4ec9c9..d49e05de5b 100644 --- a/frontend/app/view/term/termwrap.ts +++ b/frontend/app/view/term/termwrap.ts @@ -849,7 +849,7 @@ export class TermWrap { } const termSize: TermSize = { rows: this.terminal.rows, cols: this.terminal.cols }; try { - await RpcApi.ControllerInputCommand(TabRpcClient, { blockid: this.blockId, termsize: termSize }); + await RpcApi.SetTerminalSizeCommand(TabRpcClient, { blockid: this.blockId, termsize: termSize }); } catch (e) { console.log("error syncing terminal size", this.blockId, e); } diff --git a/frontend/types/gotypes.d.ts b/frontend/types/gotypes.d.ts index f81f605688..24a6ceea66 100644 --- a/frontend/types/gotypes.d.ts +++ b/frontend/types/gotypes.d.ts @@ -698,6 +698,12 @@ declare global { delete?: boolean; }; + // wshrpc.CommandSetTerminalSizeData + type CommandSetTerminalSizeData = { + blockid: string; + termsize: TermSize; + }; + // wshrpc.CommandStartBuilderData type CommandStartBuilderData = { builderid: string; diff --git a/pkg/blockcontroller/blockcontroller.go b/pkg/blockcontroller/blockcontroller.go index d7e82a02f9..93d1d86e91 100644 --- a/pkg/blockcontroller/blockcontroller.go +++ b/pkg/blockcontroller/blockcontroller.go @@ -80,6 +80,7 @@ type Controller interface { GetRuntimeStatus() *BlockControllerRuntimeStatus // does not return nil GetConnName() string SendInput(input *BlockInputUnion) error + ApplyTermSize(termSize waveobj.TermSize) error } // Registry for all controllers @@ -176,6 +177,10 @@ func ResyncController(ctx context.Context, tabId string, blockId string, rtOpts return fmt.Errorf("error getting block: %w", err) } + if rtOpts != nil && rtOpts.TermSize.Rows > 0 && rtOpts.TermSize.Cols > 0 { + SetTerminalSize(blockId, rtOpts.TermSize) + } + controllerName := blockData.Meta.GetString(waveobj.MetaKey_Controller, "") connName := blockData.Meta.GetString(waveobj.MetaKey_Connection, "") @@ -510,6 +515,53 @@ func SendInput(blockId string, inputUnion *BlockInputUnion) error { return controller.SendInput(inputUnion) } +func setTermSizeInDB(blockId string, termSize waveobj.TermSize) error { + ctx, cancelFn := context.WithTimeout(context.Background(), 2*time.Second) + defer cancelFn() + ctx = waveobj.ContextWithUpdates(ctx) + bdata, err := wstore.DBMustGet[*waveobj.Block](ctx, blockId) + if err != nil { + return fmt.Errorf("error getting block data: %v", err) + } + if bdata.RuntimeOpts == nil { + bdata.RuntimeOpts = &waveobj.RuntimeOpts{} + } + bdata.RuntimeOpts.TermSize = termSize + err = wstore.DBUpdate(ctx, bdata) + if err != nil { + return fmt.Errorf("error updating block data: %v", err) + } + updates := waveobj.ContextGetUpdatesRtn(ctx) + wps.Broker.SendUpdateEvents(updates) + return nil +} + +func SetTerminalSize(blockId string, termSize waveobj.TermSize) { + if termSize.Rows <= 0 || termSize.Cols <= 0 { + log.Printf("[blockcontroller] invalid term size for block %s: %v", blockId, termSize) + return + } + if err := setTermSizeInDB(blockId, termSize); err != nil { + log.Printf("[blockcontroller] error writing term size to DB for block %s: %v", blockId, err) + } + if c := getController(blockId); c != nil { + if err := c.ApplyTermSize(termSize); err != nil { + log.Printf("[blockcontroller] error applying term size for block %s: %v", blockId, err) + } + } +} + +func getLatestTermSize(blockId string) waveobj.TermSize { + ctx, cancelFn := context.WithTimeout(context.Background(), 2*time.Second) + defer cancelFn() + bdata, err := wstore.DBMustGet[*waveobj.Block](ctx, blockId) + if err != nil { + log.Printf("[blockcontroller] error reading block %s for term size: %v, using defaults", blockId, err) + return waveobj.TermSize{Rows: shellutil.DefaultTermRows, Cols: shellutil.DefaultTermCols} + } + return getTermSize(bdata) +} + // only call this on shutdown func StopAllBlockControllersForShutdown() { controllers := getAllControllers() diff --git a/pkg/blockcontroller/sessiondaemoncontroller.go b/pkg/blockcontroller/sessiondaemoncontroller.go index 2bff0e1dc4..d03545f51d 100644 --- a/pkg/blockcontroller/sessiondaemoncontroller.go +++ b/pkg/blockcontroller/sessiondaemoncontroller.go @@ -132,13 +132,7 @@ func (sdc *SessionDaemonController) syncJobIdToBlocks(ctx context.Context, jobId } func (sdc *SessionDaemonController) startNewJob(ctx context.Context, blockMeta waveobj.MetaMapType, rtOpts *waveobj.RuntimeOpts) (string, error) { - termSize := waveobj.TermSize{ - Rows: shellutil.DefaultTermRows, - Cols: shellutil.DefaultTermCols, - } - if rtOpts != nil && rtOpts.TermSize.Rows > 0 && rtOpts.TermSize.Cols > 0 { - termSize = rtOpts.TermSize - } + termSize := getLatestTermSize(sdc.BlockId) cmdStr := blockMeta.GetString(waveobj.MetaKey_Cmd, "") cwd := blockMeta.GetString(waveobj.MetaKey_CmdCwd, "") opts, err := remote.ParseOpts(sdc.ConnName) @@ -199,6 +193,14 @@ func (sdc *SessionDaemonController) SendInput(inputUnion *BlockInputUnion) error return daemon.SendInput(context.Background(), inputUnion.InputData, inputUnion.SigName, inputUnion.TermSize) } +func (sdc *SessionDaemonController) ApplyTermSize(termSize waveobj.TermSize) error { + daemon := sessiondaemon.Manager.Get(sdc.DaemonId) + if daemon == nil { + return nil + } + return daemon.SendInput(context.Background(), nil, "", &termSize) +} + func (sdc *SessionDaemonController) GetRuntimeStatus() *BlockControllerRuntimeStatus { var rtn BlockControllerRuntimeStatus sdc.WithLock(func() { diff --git a/pkg/blockcontroller/shellcontroller.go b/pkg/blockcontroller/shellcontroller.go index 227465b5a5..6233a82169 100644 --- a/pkg/blockcontroller/shellcontroller.go +++ b/pkg/blockcontroller/shellcontroller.go @@ -153,6 +153,16 @@ func (sc *ShellController) SendInput(inputUnion *BlockInputUnion) error { return nil } +func (sc *ShellController) ApplyTermSize(termSize waveobj.TermSize) error { + sc.Lock.Lock() + shellProc := sc.ShellProc + sc.Lock.Unlock() + if shellProc == nil { + return nil + } + return shellProc.Cmd.SetSize(termSize.Rows, termSize.Cols) +} + func (sc *ShellController) WithLock(f func()) { sc.Lock.Lock() defer sc.Lock.Unlock() @@ -303,12 +313,7 @@ func (sc *ShellController) run(logCtx context.Context, bdata *waveobj.Block, blo panichandler.PanicHandler("blockcontroller:run-shell-command", recover()) }() defer sc.UnlockRunLock() - var termSize waveobj.TermSize - if rtOpts != nil { - termSize = rtOpts.TermSize - } else { - termSize = getTermSize(bdata) - } + termSize := getLatestTermSize(sc.BlockId) err := sc.DoRunShellCommand(logCtx, &RunShellOpts{TermSize: termSize}, bdata.Meta) if err != nil { debugLog(logCtx, "error running shell: %v\n", err) @@ -569,9 +574,6 @@ func (bc *ShellController) manageRunningShellProcess(shellProc *shellexec.ShellP if len(ic.InputData) > 0 { shellProc.Cmd.Write(ic.InputData) } - if ic.TermSize != nil { - updateTermSize(shellProc, bc.BlockId, *ic.TermSize) - } } }() go func() { @@ -876,35 +878,3 @@ func getCustomInitScriptValue(meta waveobj.MetaMapType, connName string, shellTy } return "", "" } - -func updateTermSize(shellProc *shellexec.ShellProc, blockId string, termSize waveobj.TermSize) { - err := setTermSizeInDB(blockId, termSize) - if err != nil { - log.Printf("error setting pty size: %v\n", err) - } - err = shellProc.Cmd.SetSize(termSize.Rows, termSize.Cols) - if err != nil { - log.Printf("error setting pty size: %v\n", err) - } -} - -func setTermSizeInDB(blockId string, termSize waveobj.TermSize) error { - ctx, cancelFn := context.WithTimeout(context.Background(), 2*time.Second) - defer cancelFn() - ctx = waveobj.ContextWithUpdates(ctx) - bdata, err := wstore.DBMustGet[*waveobj.Block](ctx, blockId) - if err != nil { - return fmt.Errorf("error getting block data: %v", err) - } - if bdata.RuntimeOpts == nil { - bdata.RuntimeOpts = &waveobj.RuntimeOpts{} - } - bdata.RuntimeOpts.TermSize = termSize - err = wstore.DBUpdate(ctx, bdata) - if err != nil { - return fmt.Errorf("error updating block data: %v", err) - } - updates := waveobj.ContextGetUpdatesRtn(ctx) - wps.Broker.SendUpdateEvents(updates) - return nil -} diff --git a/pkg/blockcontroller/tsunamicontroller.go b/pkg/blockcontroller/tsunamicontroller.go index d064d87998..5905942ea1 100644 --- a/pkg/blockcontroller/tsunamicontroller.go +++ b/pkg/blockcontroller/tsunamicontroller.go @@ -292,6 +292,10 @@ func (c *TsunamiController) SendInput(input *BlockInputUnion) error { return fmt.Errorf("tsunami controller send input not implemented") } +func (c *TsunamiController) ApplyTermSize(termSize waveobj.TermSize) error { + return nil +} + func runTsunamiAppBinary(ctx context.Context, appBinPath string, appPath string, blockMeta waveobj.MetaMapType) (*TsunamiAppProc, error) { cmd := exec.Command(appBinPath) cmd.Env = append(os.Environ(), "TSUNAMI_CLOSEONSTDIN=1") diff --git a/pkg/wshrpc/wshclient/wshclient.go b/pkg/wshrpc/wshclient/wshclient.go index 24f7535f1a..57ec80835d 100644 --- a/pkg/wshrpc/wshclient/wshclient.go +++ b/pkg/wshrpc/wshclient/wshclient.go @@ -920,6 +920,12 @@ func SetSecretsCommand(w *wshutil.WshRpc, data map[string]*string, opts *wshrpc. return err } +// command "setterminalsize", wshserver.SetTerminalSizeCommand +func SetTerminalSizeCommand(w *wshutil.WshRpc, data wshrpc.CommandSetTerminalSizeData, opts *wshrpc.RpcOpts) error { + _, err := sendRpcRequestCallHelper[any](w, "setterminalsize", data, opts) + return err +} + // command "setvar", wshserver.SetVarCommand func SetVarCommand(w *wshutil.WshRpc, data wshrpc.CommandVarData, opts *wshrpc.RpcOpts) error { _, err := sendRpcRequestCallHelper[any](w, "setvar", data, opts) diff --git a/pkg/wshrpc/wshrpctypes.go b/pkg/wshrpc/wshrpctypes.go index fd4477a9b0..bf940b8a33 100644 --- a/pkg/wshrpc/wshrpctypes.go +++ b/pkg/wshrpc/wshrpctypes.go @@ -53,6 +53,7 @@ type WshRpcInterface interface { ControllerInputCommand(ctx context.Context, data CommandBlockInputData) error ControllerDestroyCommand(ctx context.Context, blockId string) error ControllerResyncCommand(ctx context.Context, data CommandControllerResyncData) error + SetTerminalSizeCommand(ctx context.Context, data CommandSetTerminalSizeData) error ControllerAppendOutputCommand(ctx context.Context, data CommandControllerAppendOutputData) error ResolveIdsCommand(ctx context.Context, data CommandResolveIdsData) (CommandResolveIdsRtnData, error) CreateBlockCommand(ctx context.Context, data CommandCreateBlockData) (waveobj.ORef, error) @@ -315,6 +316,11 @@ type CommandControllerResyncData struct { RtOpts *waveobj.RuntimeOpts `json:"rtopts,omitempty"` } +type CommandSetTerminalSizeData struct { + BlockId string `json:"blockid"` + TermSize waveobj.TermSize `json:"termsize"` +} + type CommandControllerAppendOutputData struct { BlockId string `json:"blockid"` Data64 string `json:"data64"` diff --git a/pkg/wshrpc/wshserver/wshserver.go b/pkg/wshrpc/wshserver/wshserver.go index 8e39b8c97c..359660c6b4 100644 --- a/pkg/wshrpc/wshserver/wshserver.go +++ b/pkg/wshrpc/wshserver/wshserver.go @@ -348,10 +348,20 @@ func (ws *WshServer) ControllerResyncCommand(ctx context.Context, data wshrpc.Co return blockcontroller.ResyncController(ctx, data.TabId, data.BlockId, data.RtOpts, data.ForceRestart) } +func (ws *WshServer) SetTerminalSizeCommand(ctx context.Context, data wshrpc.CommandSetTerminalSizeData) error { + blockcontroller.SetTerminalSize(data.BlockId, data.TermSize) + return nil +} + func (ws *WshServer) ControllerInputCommand(ctx context.Context, data wshrpc.CommandBlockInputData) error { + if data.TermSize != nil { + blockcontroller.SetTerminalSize(data.BlockId, *data.TermSize) + } + if len(data.InputData64) == 0 && data.SigName == "" { + return nil + } inputUnion := &blockcontroller.BlockInputUnion{ - SigName: data.SigName, - TermSize: data.TermSize, + SigName: data.SigName, } if len(data.InputData64) > 0 { inputBuf := make([]byte, base64.StdEncoding.DecodedLen(len(data.InputData64)))