Compare commits

..

1 Commits

Author SHA1 Message Date
bagas d029f9e93d chore: configure renovate to group and automerge dependencies
Tests / Run Tests (pull_request) Successful in 2m8s
2026-04-01 21:11:45 +07:00
10 changed files with 58 additions and 218 deletions
+4 -3
View File
@@ -13,6 +13,7 @@ import (
"tunnel_pls/internal/session/slug" "tunnel_pls/internal/session/slug"
"tunnel_pls/internal/types" "tunnel_pls/internal/types"
"tunnel_pls/internal/port"
"tunnel_pls/internal/registry" "tunnel_pls/internal/registry"
proto "git.fossy.my.id/bagas/tunnel-please-grpc/gen" proto "git.fossy.my.id/bagas/tunnel-please-grpc/gen"
@@ -884,16 +885,16 @@ func (m *mockLifecycle) Connection() ssh.Conn {
return args.Get(0).(ssh.Conn) return args.Get(0).(ssh.Conn)
} }
func (m *mockLifecycle) User() string { return m.Called().String(0) } func (m *mockLifecycle) User() string { return m.Called().String(0) }
func (m *mockLifecycle) SetChannel(channel ssh.Channel) error { return m.Called(channel).Error(0) } func (m *mockLifecycle) SetChannel(channel ssh.Channel) { m.Called(channel) }
func (m *mockLifecycle) SetStatus(status types.SessionStatus) { m.Called(status) } func (m *mockLifecycle) SetStatus(status types.SessionStatus) { m.Called(status) }
func (m *mockLifecycle) IsActive() bool { return m.Called().Bool(0) } func (m *mockLifecycle) IsActive() bool { return m.Called().Bool(0) }
func (m *mockLifecycle) StartedAt() time.Time { return m.Called().Get(0).(time.Time) } func (m *mockLifecycle) StartedAt() time.Time { return m.Called().Get(0).(time.Time) }
func (m *mockLifecycle) PortRegistry() lifecycle.PortRegistry { func (m *mockLifecycle) PortRegistry() port.Port {
args := m.Called() args := m.Called()
if args.Get(0) == nil { if args.Get(0) == nil {
return nil return nil
} }
return args.Get(0).(lifecycle.PortRegistry) return args.Get(0).(port.Port)
} }
type mockEventServiceClient struct { type mockEventServiceClient struct {
+4 -3
View File
@@ -4,6 +4,7 @@ import (
"sync" "sync"
"testing" "testing"
"time" "time"
"tunnel_pls/internal/port"
"tunnel_pls/internal/session/forwarder" "tunnel_pls/internal/session/forwarder"
"tunnel_pls/internal/session/interaction" "tunnel_pls/internal/session/interaction"
"tunnel_pls/internal/session/lifecycle" "tunnel_pls/internal/session/lifecycle"
@@ -77,15 +78,15 @@ func (ml *mockLifecycle) Connection() ssh.Conn {
return args.Get(0).(ssh.Conn) return args.Get(0).(ssh.Conn)
} }
func (ml *mockLifecycle) PortRegistry() lifecycle.PortRegistry { func (ml *mockLifecycle) PortRegistry() port.Port {
args := ml.Called() args := ml.Called()
if args.Get(0) == nil { if args.Get(0) == nil {
return nil return nil
} }
return args.Get(0).(lifecycle.PortRegistry) return args.Get(0).(port.Port)
} }
func (ml *mockLifecycle) SetChannel(channel ssh.Channel) error { return ml.Called(channel).Error(0) } func (ml *mockLifecycle) SetChannel(channel ssh.Channel) { ml.Called(channel) }
func (ml *mockLifecycle) SetStatus(status types.SessionStatus) { ml.Called(status) } func (ml *mockLifecycle) SetStatus(status types.SessionStatus) { ml.Called(status) }
func (ml *mockLifecycle) IsActive() bool { return ml.Called().Bool(0) } func (ml *mockLifecycle) IsActive() bool { return ml.Called().Bool(0) }
func (ml *mockLifecycle) StartedAt() time.Time { return ml.Called().Get(0).(time.Time) } func (ml *mockLifecycle) StartedAt() time.Time { return ml.Called().Get(0).(time.Time) }
@@ -1922,6 +1922,10 @@ func TestInteraction_Start_ProtocolSelection(t *testing.T) {
time.Sleep(50 * time.Millisecond) time.Sleep(50 * time.Millisecond)
i := mockInteraction.(*interaction) i := mockInteraction.(*interaction)
if i.program != nil {
assert.NotNil(t, i.program, "program should be initialized")
}
i.Stop() i.Stop()
mockConfig.AssertExpectations(t) mockConfig.AssertExpectations(t)
+29 -78
View File
@@ -2,7 +2,6 @@ package lifecycle
import ( import (
"errors" "errors"
"fmt"
"io" "io"
"net" "net"
"sync" "sync"
@@ -10,6 +9,8 @@ import (
"tunnel_pls/internal/session/slug" "tunnel_pls/internal/session/slug"
"tunnel_pls/internal/types" "tunnel_pls/internal/types"
portUtil "tunnel_pls/internal/port"
"golang.org/x/crypto/ssh" "golang.org/x/crypto/ssh"
) )
@@ -23,12 +24,6 @@ type SessionRegistry interface {
Remove(key types.SessionKey) Remove(key types.SessionKey)
} }
type PortRegistry interface {
Unassigned() (uint16, bool)
Claim(port uint16) bool
SetStatus(port uint16, assigned bool) error
}
type lifecycle struct { type lifecycle struct {
mu sync.Mutex mu sync.Mutex
status types.SessionStatus status types.SessionStatus
@@ -39,18 +34,18 @@ type lifecycle struct {
slug slug.Slug slug slug.Slug
startedAt time.Time startedAt time.Time
sessionRegistry SessionRegistry sessionRegistry SessionRegistry
portRegistry PortRegistry portRegistry portUtil.Port
user string user string
} }
func New(conn ssh.Conn, forwarder Forwarder, slugManager slug.Slug, port PortRegistry, sessionRegistry SessionRegistry, user string) Lifecycle { func New(conn ssh.Conn, forwarder Forwarder, slugManager slug.Slug, port portUtil.Port, sessionRegistry SessionRegistry, user string) Lifecycle {
return &lifecycle{ return &lifecycle{
status: types.SessionStatusINITIALIZING, status: types.SessionStatusINITIALIZING,
conn: conn, conn: conn,
channel: nil, channel: nil,
forwarder: forwarder, forwarder: forwarder,
slug: slugManager, slug: slugManager,
startedAt: time.Time{}, startedAt: time.Now(),
sessionRegistry: sessionRegistry, sessionRegistry: sessionRegistry,
portRegistry: port, portRegistry: port,
user: user, user: user,
@@ -60,16 +55,16 @@ func New(conn ssh.Conn, forwarder Forwarder, slugManager slug.Slug, port PortReg
type Lifecycle interface { type Lifecycle interface {
Connection() ssh.Conn Connection() ssh.Conn
Channel() ssh.Channel Channel() ssh.Channel
PortRegistry() PortRegistry PortRegistry() portUtil.Port
User() string User() string
SetChannel(channel ssh.Channel) error SetChannel(channel ssh.Channel)
SetStatus(status types.SessionStatus) SetStatus(status types.SessionStatus)
IsActive() bool IsActive() bool
StartedAt() time.Time StartedAt() time.Time
Close() error Close() error
} }
func (l *lifecycle) PortRegistry() PortRegistry { func (l *lifecycle) PortRegistry() portUtil.Port {
return l.portRegistry return l.portRegistry
} }
@@ -77,25 +72,11 @@ func (l *lifecycle) User() string {
return l.user return l.user
} }
func (l *lifecycle) SetChannel(channel ssh.Channel) error { func (l *lifecycle) SetChannel(channel ssh.Channel) {
l.mu.Lock()
defer l.mu.Unlock()
if l.status == types.SessionStatusCLOSED {
return fmt.Errorf("lifecycle is closed")
}
if channel == nil {
return fmt.Errorf("channel cannot be nil")
}
if l.channel != nil {
return fmt.Errorf("channel already set")
}
l.channel = channel l.channel = channel
return nil
} }
func (l *lifecycle) Channel() ssh.Channel { func (l *lifecycle) Channel() ssh.Channel {
l.mu.Lock()
defer l.mu.Unlock()
return l.channel return l.channel
} }
@@ -106,13 +87,7 @@ func (l *lifecycle) Connection() ssh.Conn {
func (l *lifecycle) SetStatus(status types.SessionStatus) { func (l *lifecycle) SetStatus(status types.SessionStatus) {
l.mu.Lock() l.mu.Lock()
defer l.mu.Unlock() defer l.mu.Unlock()
if l.status == types.SessionStatusCLOSED {
return
}
l.status = status l.status = status
if status == types.SessionStatusRUNNING && l.startedAt.IsZero() {
l.startedAt = time.Now()
}
} }
func (l *lifecycle) IsActive() bool { func (l *lifecycle) IsActive() bool {
@@ -123,74 +98,50 @@ func (l *lifecycle) IsActive() bool {
func (l *lifecycle) Close() error { func (l *lifecycle) Close() error {
l.mu.Lock() l.mu.Lock()
defer l.mu.Unlock()
if l.status == types.SessionStatusCLOSED { if l.status == types.SessionStatusCLOSED {
closeErr := l.closeErr return l.closeErr
l.mu.Unlock()
return closeErr
} }
l.status = types.SessionStatusCLOSED l.status = types.SessionStatusCLOSED
channel := l.channel
conn := l.conn
l.mu.Unlock()
var errs []error var errs []error
if channel != nil { tunnelType := l.forwarder.TunnelType()
if err := channel.Close(); err != nil && !isClosedError(err) {
errs = append(errs, err) if l.channel != nil {
} if err := l.channel.Close(); err != nil && !isClosedError(err) {
}
if conn != nil {
if err := conn.Close(); err != nil && !isClosedError(err) {
errs = append(errs, err) errs = append(errs, err)
} }
} }
l.cleanupRegistry() if l.conn != nil {
if err := l.cleanupForwarder(); err != nil { if err := l.conn.Close(); err != nil && !isClosedError(err) {
errs = append(errs, err) errs = append(errs, err)
}
} }
closeErr := errors.Join(errs...) clientSlug := l.slug.String()
l.mu.Lock()
l.closeErr = closeErr
l.mu.Unlock()
return closeErr
}
func (l *lifecycle) cleanupRegistry() {
slugStr := l.slug.String()
if slugStr == "" {
return
}
key := types.SessionKey{ key := types.SessionKey{
Id: slugStr, Id: clientSlug,
Type: l.forwarder.TunnelType(), Type: tunnelType,
} }
l.sessionRegistry.Remove(key) l.sessionRegistry.Remove(key)
}
func (l *lifecycle) cleanupForwarder() error { if tunnelType == types.TunnelTypeTCP {
if l.forwarder.TunnelType() != types.TunnelTypeTCP { errs = append(errs, l.PortRegistry().SetStatus(l.forwarder.ForwardedPort(), false))
return nil errs = append(errs, l.forwarder.Close())
} }
var errs []error
errs = append(errs, l.portRegistry.SetStatus(l.forwarder.ForwardedPort(), false)) l.closeErr = errors.Join(errs...)
errs = append(errs, l.forwarder.Close()) return l.closeErr
return errors.Join(errs...)
} }
func isClosedError(err error) bool { func isClosedError(err error) bool {
if err == nil { if err == nil {
return false return false
} }
return errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) return errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) || err.Error() == "EOF"
} }
func (l *lifecycle) StartedAt() time.Time { func (l *lifecycle) StartedAt() time.Time {
l.mu.Lock()
defer l.mu.Unlock()
return l.startedAt return l.startedAt
} }
+4 -125
View File
@@ -5,7 +5,6 @@ import (
"errors" "errors"
"io" "io"
"net" "net"
"sync"
"testing" "testing"
"tunnel_pls/internal/types" "tunnel_pls/internal/types"
@@ -178,14 +177,8 @@ func TestLifecycle_SetChannel(t *testing.T) {
mockSSHChannel := &MockSSHChannel{} mockSSHChannel := &MockSSHChannel{}
err := mockLifecycle.SetChannel(mockSSHChannel) mockLifecycle.SetChannel(mockSSHChannel)
assert.NoError(t, err)
assert.Equal(t, mockSSHChannel, mockLifecycle.Channel())
anotherChannel := &MockSSHChannel{}
err = mockLifecycle.SetChannel(anotherChannel)
assert.Error(t, err)
assert.Contains(t, err.Error(), "channel already set")
assert.Equal(t, mockSSHChannel, mockLifecycle.Channel()) assert.Equal(t, mockSSHChannel, mockLifecycle.Channel())
} }
@@ -283,15 +276,14 @@ func TestLifecycle_Close(t *testing.T) {
mockLifecycle := New(mockSSHConn, mockForwarder, mockSlug, mockPort, mockSessionRegistry, "mas-fuad") mockLifecycle := New(mockSSHConn, mockForwarder, mockSlug, mockPort, mockSessionRegistry, "mas-fuad")
mockLifecycle.SetStatus(types.SessionStatusRUNNING) mockLifecycle.SetStatus(types.SessionStatusRUNNING)
err := mockLifecycle.SetChannel(mockSSHChannel) mockLifecycle.SetChannel(mockSSHChannel)
assert.NoError(t, err)
if tt.alreadyClosed { if tt.alreadyClosed {
err = mockLifecycle.Close() err := mockLifecycle.Close()
assert.NoError(t, err) assert.NoError(t, err)
} }
err = mockLifecycle.Close() err := mockLifecycle.Close()
if tt.expectErr { if tt.expectErr {
assert.Error(t, err) assert.Error(t, err)
@@ -309,116 +301,3 @@ func TestLifecycle_Close(t *testing.T) {
}) })
} }
} }
func TestLifecycle_ConcurrentClose(t *testing.T) {
mockSSHConn := &MockSSHConn{}
mockSSHConn.On("Close").Return(nil)
mockForwarder := &MockForwarder{}
mockForwarder.On("TunnelType").Return(types.TunnelTypeHTTP)
mockSlug := &MockSlug{}
mockSlug.On("String").Return("test-slug")
mockPort := &MockPort{}
mockSessionRegistry := &MockSessionRegistry{}
mockSessionRegistry.On("Remove", mock.Anything).Return()
mockSSHChannel := &MockSSHChannel{}
mockSSHChannel.On("Close").Return(nil)
mockLifecycle := New(mockSSHConn, mockForwarder, mockSlug, mockPort, mockSessionRegistry, "mas-fuad")
mockLifecycle.SetStatus(types.SessionStatusRUNNING)
err := mockLifecycle.SetChannel(mockSSHChannel)
assert.NoError(t, err)
const numGoroutines = 10
var wg sync.WaitGroup
errChan := make(chan error, numGoroutines)
for i := 0; i < numGoroutines; i++ {
wg.Add(1)
go func() {
defer wg.Done()
err := mockLifecycle.Close()
errChan <- err
}()
}
wg.Wait()
close(errChan)
for err := range errChan {
assert.NoError(t, err)
}
assert.False(t, mockLifecycle.IsActive())
}
func TestLifecycle_SetChannel_AfterClose(t *testing.T) {
mockSSHConn := new(MockSSHConn)
mockSSHConn.On("Close").Return(nil)
mockForwarder := &MockForwarder{}
mockForwarder.On("TunnelType").Return(types.TunnelTypeHTTP)
mockSlug := &MockSlug{}
mockSlug.On("String").Return("test-slug")
mockPort := &MockPort{}
mockSessionRegistry := &MockSessionRegistry{}
mockSessionRegistry.On("Remove", mock.Anything).Return()
mockSSHChannel := &MockSSHChannel{}
mockSSHChannel.On("Close").Return(nil)
mockLifecycle := New(mockSSHConn, mockForwarder, mockSlug, mockPort, mockSessionRegistry, "mas-fuad")
mockLifecycle.SetStatus(types.SessionStatusRUNNING)
err := mockLifecycle.SetChannel(mockSSHChannel)
assert.NoError(t, err)
err = mockLifecycle.Close()
assert.NoError(t, err)
anotherChannel := &MockSSHChannel{}
err = mockLifecycle.SetChannel(anotherChannel)
assert.Error(t, err)
assert.Contains(t, err.Error(), "lifecycle is closed")
}
func TestLifecycle_SetChannel_Nil(t *testing.T) {
mockSSHConn := new(MockSSHConn)
mockForwarder := &MockForwarder{}
mockSlug := &MockSlug{}
mockPort := &MockPort{}
mockSessionRegistry := &MockSessionRegistry{}
mockLifecycle := New(mockSSHConn, mockForwarder, mockSlug, mockPort, mockSessionRegistry, "mas-fuad")
err := mockLifecycle.SetChannel(nil)
assert.Error(t, err)
assert.Contains(t, err.Error(), "channel cannot be nil")
}
func TestLifecycle_SetStatus_AfterClose(t *testing.T) {
mockSSHConn := new(MockSSHConn)
mockSSHConn.On("Close").Return(nil)
mockForwarder := &MockForwarder{}
mockForwarder.On("TunnelType").Return(types.TunnelTypeHTTP)
mockSlug := &MockSlug{}
mockSlug.On("String").Return("test-slug")
mockPort := &MockPort{}
mockSessionRegistry := &MockSessionRegistry{}
mockSessionRegistry.On("Remove", mock.Anything).Return()
mockSSHChannel := &MockSSHChannel{}
mockSSHChannel.On("Close").Return(nil)
mockLifecycle := New(mockSSHConn, mockForwarder, mockSlug, mockPort, mockSessionRegistry, "mas-fuad")
mockLifecycle.SetStatus(types.SessionStatusRUNNING)
err := mockLifecycle.SetChannel(mockSSHChannel)
assert.NoError(t, err)
err = mockLifecycle.Close()
assert.NoError(t, err)
assert.False(t, mockLifecycle.IsActive())
mockLifecycle.SetStatus(types.SessionStatusRUNNING)
assert.False(t, mockLifecycle.IsActive(), "SetStatus should be ignored after Close")
}
+1 -3
View File
@@ -164,9 +164,7 @@ func (s *session) setupInteractiveMode(channel ssh.NewChannel) error {
} }
}() }()
if err = s.lifecycle.SetChannel(ch); err != nil { s.lifecycle.SetChannel(ch)
return err
}
s.interaction.SetChannel(ch) s.interaction.SetChannel(ch)
s.interaction.SetMode(types.InteractiveModeINTERACTIVE) s.interaction.SetMode(types.InteractiveModeINTERACTIVE)
+2 -1
View File
@@ -55,7 +55,8 @@ func TestHTTPServer_Serve(t *testing.T) {
go func() { go func() {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
_ = listener.Close() err = listener.Close()
assert.NoError(t, err)
}() }()
err = srv.Serve(listener) err = srv.Serve(listener)
+2 -1
View File
@@ -63,7 +63,8 @@ func TestHTTPSServer_Serve(t *testing.T) {
go func() { go func() {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
_ = listener.Close() err = listener.Close()
assert.NoError(t, err)
}() }()
err = srv.Serve(listener) err = srv.Serve(listener)
+2 -1
View File
@@ -45,7 +45,8 @@ func TestTCPServer_Serve(t *testing.T) {
go func() { go func() {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
_ = listener.Close() err = listener.Close()
assert.NoError(t, err)
}() }()
err = srv.Serve(listener) err = srv.Serve(listener)
+5 -2
View File
@@ -2,6 +2,8 @@
"extends": [ "extends": [
"config:recommended" "config:recommended"
], ],
"prConcurrentLimit": 1,
"prHourlyLimit": 1,
"packageRules": [ "packageRules": [
{ {
"matchUpdateTypes": [ "matchUpdateTypes": [
@@ -10,9 +12,10 @@
"pin", "pin",
"digest" "digest"
], ],
"groupName": "all-dependencies",
"automerge": true, "automerge": true,
"baseBranchPatterns": [ "matchPackageNames": [
"staging" "*"
] ]
} }
] ]