fix the memory bug

This commit is contained in:
Kujtim Hoxha
2025-04-21 13:33:51 +02:00
parent 1da298e755
commit e7bb99baab
6 changed files with 128 additions and 56 deletions

View File

@@ -351,9 +351,12 @@ go build -o opencode
## Acknowledgments ## Acknowledgments
OpenCode builds upon the work of several open source projects and developers: OpenCode gratefully acknowledges the contributions and support from these key individuals:
- [@isaacphi](https://github.com/isaacphi) - LSP client implementation - [@isaacphi](https://github.com/isaacphi) - For the [mcp-language-server](https://github.com/isaacphi/mcp-language-server) project which provided the foundation for our LSP client implementation
- [@adamdottv](https://github.com/adamdottv) - For the design direction and UI/UX architecture
Special thanks to the broader open source community whose tools and libraries have made this project possible.
## License ## License

View File

@@ -79,7 +79,7 @@ var rootCmd = &cobra.Command{
initMCPTools(ctx, app) initMCPTools(ctx, app)
// Setup the subscriptions, this will send services events to the TUI // Setup the subscriptions, this will send services events to the TUI
ch, cancelSubs := setupSubscriptions(app) ch, cancelSubs := setupSubscriptions(app, ctx)
// Create a context for the TUI message handler // Create a context for the TUI message handler
tuiCtx, tuiCancel := context.WithCancel(ctx) tuiCtx, tuiCancel := context.WithCancel(ctx)
@@ -174,21 +174,21 @@ func setupSubscriber[T any](
defer wg.Done() defer wg.Done()
defer logging.RecoverPanic(fmt.Sprintf("subscription-%s", name), nil) defer logging.RecoverPanic(fmt.Sprintf("subscription-%s", name), nil)
subCh := subscriber(ctx)
for { for {
select { select {
case event, ok := <-subscriber(ctx): case event, ok := <-subCh:
if !ok { if !ok {
logging.Info("%s subscription channel closed", name) logging.Info("%s subscription channel closed", name)
return return
} }
// Convert generic event to tea.Msg if needed
var msg tea.Msg = event var msg tea.Msg = event
// Non-blocking send with timeout to prevent deadlocks
select { select {
case outputCh <- msg: case outputCh <- msg:
case <-time.After(500 * time.Millisecond): case <-time.After(2 * time.Second):
logging.Warn("%s message dropped due to slow consumer", name) logging.Warn("%s message dropped due to slow consumer", name)
case <-ctx.Done(): case <-ctx.Done():
logging.Info("%s subscription cancelled", name) logging.Info("%s subscription cancelled", name)
@@ -202,23 +202,21 @@ func setupSubscriber[T any](
}() }()
} }
func setupSubscriptions(app *app.App) (chan tea.Msg, func()) { func setupSubscriptions(app *app.App, parentCtx context.Context) (chan tea.Msg, func()) {
ch := make(chan tea.Msg, 100) ch := make(chan tea.Msg, 100)
// Add a buffer to prevent blocking
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(parentCtx) // Inherit from parent context
// Setup each subscription using the helper
setupSubscriber(ctx, &wg, "logging", logging.Subscribe, ch) setupSubscriber(ctx, &wg, "logging", logging.Subscribe, ch)
setupSubscriber(ctx, &wg, "sessions", app.Sessions.Subscribe, ch) setupSubscriber(ctx, &wg, "sessions", app.Sessions.Subscribe, ch)
setupSubscriber(ctx, &wg, "messages", app.Messages.Subscribe, ch) setupSubscriber(ctx, &wg, "messages", app.Messages.Subscribe, ch)
setupSubscriber(ctx, &wg, "permissions", app.Permissions.Subscribe, ch) setupSubscriber(ctx, &wg, "permissions", app.Permissions.Subscribe, ch)
// Return channel and a cleanup function
cleanupFunc := func() { cleanupFunc := func() {
logging.Info("Cancelling all subscriptions") logging.Info("Cancelling all subscriptions")
cancel() // Signal all goroutines to stop cancel() // Signal all goroutines to stop
// Wait with a timeout for all goroutines to complete
waitCh := make(chan struct{}) waitCh := make(chan struct{})
go func() { go func() {
defer logging.RecoverPanic("subscription-cleanup", nil) defer logging.RecoverPanic("subscription-cleanup", nil)
@@ -229,11 +227,11 @@ func setupSubscriptions(app *app.App) (chan tea.Msg, func()) {
select { select {
case <-waitCh: case <-waitCh:
logging.Info("All subscription goroutines completed successfully") logging.Info("All subscription goroutines completed successfully")
close(ch) // Only close after all writers are confirmed done
case <-time.After(5 * time.Second): case <-time.After(5 * time.Second):
logging.Warn("Timed out waiting for some subscription goroutines to complete") logging.Warn("Timed out waiting for some subscription goroutines to complete")
close(ch)
} }
close(ch) // Safe to close after all writers are done or timed out
} }
return ch, cleanupFunc return ch, cleanupFunc
} }

View File

@@ -5,47 +5,53 @@ import (
"sync" "sync"
) )
const bufferSize = 1024 const bufferSize = 64
// Broker allows clients to publish events and subscribe to events
type Broker[T any] struct { type Broker[T any] struct {
subs map[chan Event[T]]struct{} // subscriptions subs map[chan Event[T]]struct{}
mu sync.Mutex // sync access to map mu sync.RWMutex
done chan struct{} // close when broker is shutting down done chan struct{}
subCount int
maxEvents int
} }
// NewBroker constructs a pub/sub broker.
func NewBroker[T any]() *Broker[T] { func NewBroker[T any]() *Broker[T] {
return NewBrokerWithOptions[T](bufferSize, 1000)
}
func NewBrokerWithOptions[T any](channelBufferSize, maxEvents int) *Broker[T] {
b := &Broker[T]{ b := &Broker[T]{
subs: make(map[chan Event[T]]struct{}), subs: make(map[chan Event[T]]struct{}),
done: make(chan struct{}), done: make(chan struct{}),
subCount: 0,
maxEvents: maxEvents,
} }
return b return b
} }
// Shutdown the broker, terminating any subscriptions.
func (b *Broker[T]) Shutdown() { func (b *Broker[T]) Shutdown() {
close(b.done) select {
case <-b.done: // Already closed
return
default:
close(b.done)
}
b.mu.Lock() b.mu.Lock()
defer b.mu.Unlock() defer b.mu.Unlock()
// Remove each subscriber entry, so Publish() cannot send any further
// messages, and close each subscriber's channel, so the subscriber cannot
// consume any more messages.
for ch := range b.subs { for ch := range b.subs {
delete(b.subs, ch) delete(b.subs, ch)
close(ch) close(ch)
} }
b.subCount = 0
} }
// Subscribe subscribes the caller to a stream of events. The returned channel
// is closed when the broker is shutdown.
func (b *Broker[T]) Subscribe(ctx context.Context) <-chan Event[T] { func (b *Broker[T]) Subscribe(ctx context.Context) <-chan Event[T] {
b.mu.Lock() b.mu.Lock()
defer b.mu.Unlock() defer b.mu.Unlock()
// Check if broker has shutdown and if so return closed channel
select { select {
case <-b.done: case <-b.done:
ch := make(chan Event[T]) ch := make(chan Event[T])
@@ -54,18 +60,16 @@ func (b *Broker[T]) Subscribe(ctx context.Context) <-chan Event[T] {
default: default:
} }
// Subscribe
sub := make(chan Event[T], bufferSize) sub := make(chan Event[T], bufferSize)
b.subs[sub] = struct{}{} b.subs[sub] = struct{}{}
b.subCount++
// Unsubscribe when context is done.
go func() { go func() {
<-ctx.Done() <-ctx.Done()
b.mu.Lock() b.mu.Lock()
defer b.mu.Unlock() defer b.mu.Unlock()
// Check if broker has shutdown and if so do nothing
select { select {
case <-b.done: case <-b.done:
return return
@@ -74,21 +78,39 @@ func (b *Broker[T]) Subscribe(ctx context.Context) <-chan Event[T] {
delete(b.subs, sub) delete(b.subs, sub)
close(sub) close(sub)
b.subCount--
}() }()
return sub return sub
} }
// Publish an event to subscribers. func (b *Broker[T]) GetSubscriberCount() int {
func (b *Broker[T]) Publish(t EventType, payload T) { b.mu.RLock()
b.mu.Lock() defer b.mu.RUnlock()
defer b.mu.Unlock() return b.subCount
}
func (b *Broker[T]) Publish(t EventType, payload T) {
b.mu.RLock()
select {
case <-b.done:
b.mu.RUnlock()
return
default:
}
subscribers := make([]chan Event[T], 0, len(b.subs))
for sub := range b.subs { for sub := range b.subs {
subscribers = append(subscribers, sub)
}
b.mu.RUnlock()
event := Event[T]{Type: t, Payload: payload}
for _, sub := range subscribers {
select { select {
case sub <- Event[T]{Type: t, Payload: payload}: case sub <- event:
case <-b.done: default:
return
} }
} }
} }

View File

@@ -370,6 +370,7 @@ func (m *messagesCmp) SetSize(width, height int) tea.Cmd {
delete(m.cachedContent, msg.ID) delete(m.cachedContent, msg.ID)
} }
m.uiMessages = make([]uiMessage, 0) m.uiMessages = make([]uiMessage, 0)
m.renderView()
return nil return nil
} }

View File

@@ -18,6 +18,11 @@ import (
"github.com/kujtimiihoxha/opencode/internal/tui/util" "github.com/kujtimiihoxha/opencode/internal/tui/util"
) )
type StatusCmp interface {
tea.Model
SetHelpMsg(string)
}
type statusCmp struct { type statusCmp struct {
info util.InfoMsg info util.InfoMsg
width int width int
@@ -146,7 +151,7 @@ func (m *statusCmp) projectDiagnostics() string {
break break
} }
} }
// If any server is initializing, show that status // If any server is initializing, show that status
if initializing { if initializing {
return lipgloss.NewStyle(). return lipgloss.NewStyle().
@@ -154,7 +159,7 @@ func (m *statusCmp) projectDiagnostics() string {
Foreground(styles.Peach). Foreground(styles.Peach).
Render(fmt.Sprintf("%s Initializing LSP...", styles.SpinnerIcon)) Render(fmt.Sprintf("%s Initializing LSP...", styles.SpinnerIcon))
} }
errorDiagnostics := []protocol.Diagnostic{} errorDiagnostics := []protocol.Diagnostic{}
warnDiagnostics := []protocol.Diagnostic{} warnDiagnostics := []protocol.Diagnostic{}
hintDiagnostics := []protocol.Diagnostic{} hintDiagnostics := []protocol.Diagnostic{}
@@ -235,7 +240,11 @@ func (m statusCmp) model() string {
return styles.Padded.Background(styles.Grey).Foreground(styles.Text).Render(model.Name) return styles.Padded.Background(styles.Grey).Foreground(styles.Text).Render(model.Name)
} }
func NewStatusCmp(lspClients map[string]*lsp.Client) tea.Model { func (m statusCmp) SetHelpMsg(s string) {
helpWidget = styles.Padded.Background(styles.Forground).Foreground(styles.BackgroundDarker).Bold(true).Render(s)
}
func NewStatusCmp(lspClients map[string]*lsp.Client) StatusCmp {
return &statusCmp{ return &statusCmp{
messageTTL: 10 * time.Second, messageTTL: 10 * time.Second,
lspClients: lspClients, lspClients: lspClients,

View File

@@ -39,12 +39,18 @@ var keys = keyMap{
key.WithKeys("ctrl+_"), key.WithKeys("ctrl+_"),
key.WithHelp("ctrl+?", "toggle help"), key.WithHelp("ctrl+?", "toggle help"),
), ),
SwitchSession: key.NewBinding( SwitchSession: key.NewBinding(
key.WithKeys("ctrl+a"), key.WithKeys("ctrl+a"),
key.WithHelp("ctrl+a", "switch session"), key.WithHelp("ctrl+a", "switch session"),
), ),
} }
var helpEsc = key.NewBinding(
key.WithKeys("?"),
key.WithHelp("?", "toggle help"),
)
var returnKey = key.NewBinding( var returnKey = key.NewBinding(
key.WithKeys("esc"), key.WithKeys("esc"),
key.WithHelp("esc", "close"), key.WithHelp("esc", "close"),
@@ -61,7 +67,7 @@ type appModel struct {
previousPage page.PageID previousPage page.PageID
pages map[page.PageID]tea.Model pages map[page.PageID]tea.Model
loadedPages map[page.PageID]bool loadedPages map[page.PageID]bool
status tea.Model status core.StatusCmp
app *app.App app *app.App
showPermissions bool showPermissions bool
@@ -75,6 +81,8 @@ type appModel struct {
showSessionDialog bool showSessionDialog bool
sessionDialog dialog.SessionDialog sessionDialog dialog.SessionDialog
editingMode bool
} }
func (a appModel) Init() tea.Cmd { func (a appModel) Init() tea.Cmd {
@@ -101,7 +109,8 @@ func (a appModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
msg.Height -= 1 // Make space for the status bar msg.Height -= 1 // Make space for the status bar
a.width, a.height = msg.Width, msg.Height a.width, a.height = msg.Width, msg.Height
a.status, _ = a.status.Update(msg) s, _ := a.status.Update(msg)
a.status = s.(core.StatusCmp)
a.pages[a.currentPage], cmd = a.pages[a.currentPage].Update(msg) a.pages[a.currentPage], cmd = a.pages[a.currentPage].Update(msg)
cmds = append(cmds, cmd) cmds = append(cmds, cmd)
@@ -118,45 +127,56 @@ func (a appModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
cmds = append(cmds, sessionCmd) cmds = append(cmds, sessionCmd)
return a, tea.Batch(cmds...) return a, tea.Batch(cmds...)
case chat.EditorFocusMsg:
a.editingMode = bool(msg)
// Status // Status
case util.InfoMsg: case util.InfoMsg:
a.status, cmd = a.status.Update(msg) s, cmd := a.status.Update(msg)
a.status = s.(core.StatusCmp)
cmds = append(cmds, cmd) cmds = append(cmds, cmd)
return a, tea.Batch(cmds...) return a, tea.Batch(cmds...)
case pubsub.Event[logging.LogMessage]: case pubsub.Event[logging.LogMessage]:
if msg.Payload.Persist { if msg.Payload.Persist {
switch msg.Payload.Level { switch msg.Payload.Level {
case "error": case "error":
a.status, cmd = a.status.Update(util.InfoMsg{ s, cmd := a.status.Update(util.InfoMsg{
Type: util.InfoTypeError, Type: util.InfoTypeError,
Msg: msg.Payload.Message, Msg: msg.Payload.Message,
TTL: msg.Payload.PersistTime, TTL: msg.Payload.PersistTime,
}) })
a.status = s.(core.StatusCmp)
cmds = append(cmds, cmd)
case "info": case "info":
a.status, cmd = a.status.Update(util.InfoMsg{ s, cmd := a.status.Update(util.InfoMsg{
Type: util.InfoTypeInfo, Type: util.InfoTypeInfo,
Msg: msg.Payload.Message, Msg: msg.Payload.Message,
TTL: msg.Payload.PersistTime, TTL: msg.Payload.PersistTime,
}) })
a.status = s.(core.StatusCmp)
cmds = append(cmds, cmd)
case "warn": case "warn":
a.status, cmd = a.status.Update(util.InfoMsg{ s, cmd := a.status.Update(util.InfoMsg{
Type: util.InfoTypeWarn, Type: util.InfoTypeWarn,
Msg: msg.Payload.Message, Msg: msg.Payload.Message,
TTL: msg.Payload.PersistTime, TTL: msg.Payload.PersistTime,
}) })
a.status = s.(core.StatusCmp)
cmds = append(cmds, cmd)
default: default:
a.status, cmd = a.status.Update(util.InfoMsg{ s, cmd := a.status.Update(util.InfoMsg{
Type: util.InfoTypeInfo, Type: util.InfoTypeInfo,
Msg: msg.Payload.Message, Msg: msg.Payload.Message,
TTL: msg.Payload.PersistTime, TTL: msg.Payload.PersistTime,
}) })
a.status = s.(core.StatusCmp)
cmds = append(cmds, cmd)
} }
cmds = append(cmds, cmd)
} }
case util.ClearStatusMsg: case util.ClearStatusMsg:
a.status, _ = a.status.Update(msg) s, _ := a.status.Update(msg)
a.status = s.(core.StatusCmp)
// Permission // Permission
case pubsub.Event[permission.PermissionRequest]: case pubsub.Event[permission.PermissionRequest]:
@@ -243,7 +263,16 @@ func (a appModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
} }
a.showHelp = !a.showHelp a.showHelp = !a.showHelp
return a, nil return a, nil
case key.Matches(msg, helpEsc):
if !a.editingMode {
if a.showQuit {
return a, nil
}
a.showHelp = !a.showHelp
return a, nil
}
} }
} }
if a.showQuit { if a.showQuit {
@@ -275,7 +304,8 @@ func (a appModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
} }
} }
a.status, _ = a.status.Update(msg) s, _ := a.status.Update(msg)
a.status = s.(core.StatusCmp)
a.pages[a.currentPage], cmd = a.pages[a.currentPage].Update(msg) a.pages[a.currentPage], cmd = a.pages[a.currentPage].Update(msg)
cmds = append(cmds, cmd) cmds = append(cmds, cmd)
return a, tea.Batch(cmds...) return a, tea.Batch(cmds...)
@@ -326,6 +356,12 @@ func (a appModel) View() string {
) )
} }
if a.editingMode {
a.status.SetHelpMsg("ctrl+? help")
} else {
a.status.SetHelpMsg("? help")
}
if a.showHelp { if a.showHelp {
bindings := layout.KeyMapToSlice(keys) bindings := layout.KeyMapToSlice(keys)
if p, ok := a.pages[a.currentPage].(layout.Bindings); ok { if p, ok := a.pages[a.currentPage].(layout.Bindings); ok {
@@ -337,7 +373,9 @@ func (a appModel) View() string {
if a.currentPage == page.LogsPage { if a.currentPage == page.LogsPage {
bindings = append(bindings, logsKeyReturnKey) bindings = append(bindings, logsKeyReturnKey)
} }
if !a.editingMode {
bindings = append(bindings, helpEsc)
}
a.help.SetBindings(bindings) a.help.SetBindings(bindings)
overlay := a.help.View() overlay := a.help.View()
@@ -398,6 +436,7 @@ func New(app *app.App) tea.Model {
sessionDialog: dialog.NewSessionDialogCmp(), sessionDialog: dialog.NewSessionDialogCmp(),
permissions: dialog.NewPermissionDialogCmp(), permissions: dialog.NewPermissionDialogCmp(),
app: app, app: app,
editingMode: true,
pages: map[page.PageID]tea.Model{ pages: map[page.PageID]tea.Model{
page.ChatPage: page.NewChatPage(app), page.ChatPage: page.NewChatPage(app),
page.LogsPage: page.NewLogsPage(), page.LogsPage: page.NewLogsPage(),