Compare commits

..

8 Commits

Author SHA1 Message Date
bagas 198147c676 fix: race condition for err variable at session package
Tests / Run Tests (pull_request) Successful in 3m10s
Docker Build and Push / Run Tests (push) Successful in 3m16s
Docker Build and Push / Build and Push Docker Image (push) Successful in 15m8s
2026-07-16 20:44:45 +07:00
bagas 1c591882cc fix: make slug package thread safe 2026-07-16 20:35:10 +07:00
bagas d9f3737797 fix: make forwarder package thread safe 2026-07-16 19:55:57 +07:00
bagas 3c6e44cd0a fix: wait for initial payload 2026-07-16 19:23:09 +07:00
bagas e8be839dac Fix race conditions and improve lifecycle safety (#150)
SonarQube Scan / SonarQube Trigger (push) Successful in 4m1s
Docker Build and Push / Run Tests (push) Successful in 2m28s
Docker Build and Push / Build and Push Docker Image (push) Successful in 18m27s
Tests / Run Tests (pull_request) Successful in 2m29s
## Summary
Comprehensive fixes for race conditions, concurrency bugs, and code quality improvements in the lifecycle module and related test files.

## Changes

### Race Condition Fixes
- Fixed data race in `SetChannel()` and `Channel()` by adding mutex protection
- Fixed data race in `StartedAt()` by adding mutex protection
- Fixed test race conditions in `transport` and `interaction` packages where variables were shared between goroutines

### Concurrency Safety Improvements
- Refactored `Close()` to release mutex before calling external code, preventing potential deadlocks
- Made `SetChannel()` reject calls after `Close()` to prevent setting channel on torn-down lifecycle
- Made `SetStatus()` ignore calls after `Close()` to ensure CLOSED is a terminal state
- Added nil check to `SetChannel()` to prevent masking upstream bugs

### Code Quality Improvements
- Refactored `Close()` to extract helper methods: `closeChannel()`, `closeConnection()`, `cleanupRegistry()`, `cleanupForwarder()`
- Fixed `isClosedError()` to remove redundant string comparison
- Added `PortRegistry` interface for better interface segregation (removed dependency on full `port.Port`)
- Made `cleanupRegistry()` conditional on non-empty slug to avoid unnecessary registry calls
- Fixed `startedAt` to be set when session starts running instead of at object creation

Reviewed-on: #150
Co-authored-by: Bagas <bagas@fossy.my.id>
Co-committed-by: Bagas <bagas@fossy.my.id>
2026-07-16 17:31:14 +07:00
bagas d7ec5e852b Merge pull request 'refactor: move types package to internal' (#122) from refactor/move-types-to-internal into staging
SonarQube Scan / SonarQube Trigger (push) Successful in 3m32s
Tests / Run Tests (pull_request) Successful in 2m6s
Reviewed-on: #122
2026-03-30 11:53:16 +07:00
bagas dc0a4af27f Merge pull request 'refactor: move session package to internal' (#121) from refactor/move-session-to-internal into staging
SonarQube Scan / SonarQube Trigger (push) Successful in 3m48s
Reviewed-on: #121
2026-03-30 11:47:35 +07:00
bagas 032b6e5096 Merge pull request 'refactor: move server package to internal' (#120) from refactor/move-server-to-internal into staging
SonarQube Scan / SonarQube Trigger (push) Successful in 3m36s
Reviewed-on: #120
2026-03-30 11:38:52 +07:00
14 changed files with 379 additions and 257 deletions
+66 -177
View File
@@ -9,6 +9,7 @@ jobs:
test:
name: Run Tests
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v6
@@ -28,168 +29,15 @@ jobs:
- name: Run tests
run: go test -v -p 4 ./...
build-amd64:
name: Build (linux/amd64)
runs-on: [ubuntu-latest, amd64]
needs: test
steps:
- name: Checkout repository
uses: actions/checkout@v6
- name: Extract version
id: version
run: |
VERSION=${GITHUB_REF#refs/tags/v}
echo "VERSION=$VERSION" >> $GITHUB_OUTPUT
echo "BUILD_DATE=$(date -u +'%Y-%m-%dT%H:%M:%SZ')" >> $GITHUB_OUTPUT
echo "COMMIT=${{ github.sha }}" >> $GITHUB_OUTPUT
if echo "$VERSION" | grep -qE '^[0-9]+\.[0-9]+\.[0-9]+(-[a-zA-Z0-9.]+)?$'; then
echo "MAJOR=$(echo "$VERSION" | cut -d. -f1)" >> $GITHUB_OUTPUT
echo "MINOR=$(echo "$VERSION" | cut -d. -f2)" >> $GITHUB_OUTPUT
if echo "$VERSION" | grep -q '-'; then
echo "IS_PRERELEASE=true" >> $GITHUB_OUTPUT
else
echo "IS_PRERELEASE=false" >> $GITHUB_OUTPUT
fi
else
echo "Invalid version format: $VERSION"
exit 1
fi
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log in to Docker Registry
uses: docker/login-action@v3
with:
registry: git.fossy.my.id
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Build and push by digest
id: build
uses: docker/build-push-action@v6
with:
context: .
platforms: linux/amd64
outputs: >-
type=image,name=git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please,push-by-digest=true,name-canonical=true,push=true
build-args: |
VERSION=${{ steps.version.outputs.VERSION }}
BUILD_DATE=${{ steps.version.outputs.BUILD_DATE }}
COMMIT=${{ steps.version.outputs.COMMIT }}
- name: Export digest
run: |
mkdir -p /tmp/digests
digest="${{ steps.build.outputs.digest }}"
touch "/tmp/digests/${digest#sha256:}"
- name: Upload digest
uses: actions/upload-artifact@v7
with:
name: digests-linux-amd64
path: /tmp/digests/*
if-no-files-found: error
retention-days: 1
build-arm64:
name: Build (linux/arm64)
runs-on: [ubuntu-latest, arm64]
needs: test
steps:
- name: Checkout repository
uses: actions/checkout@v6
- name: Extract version
id: version
run: |
VERSION=${GITHUB_REF#refs/tags/v}
echo "VERSION=$VERSION" >> $GITHUB_OUTPUT
echo "BUILD_DATE=$(date -u +'%Y-%m-%dT%H:%M:%SZ')" >> $GITHUB_OUTPUT
echo "COMMIT=${{ github.sha }}" >> $GITHUB_OUTPUT
if echo "$VERSION" | grep -qE '^[0-9]+\.[0-9]+\.[0-9]+(-[a-zA-Z0-9.]+)?$'; then
echo "MAJOR=$(echo "$VERSION" | cut -d. -f1)" >> $GITHUB_OUTPUT
echo "MINOR=$(echo "$VERSION" | cut -d. -f2)" >> $GITHUB_OUTPUT
if echo "$VERSION" | grep -q '-'; then
echo "IS_PRERELEASE=true" >> $GITHUB_OUTPUT
else
echo "IS_PRERELEASE=false" >> $GITHUB_OUTPUT
fi
else
echo "Invalid version format: $VERSION"
exit 1
fi
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log in to Docker Registry
uses: docker/login-action@v3
with:
registry: git.fossy.my.id
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Build and push by digest
id: build
uses: docker/build-push-action@v6
with:
context: .
platforms: linux/arm64
outputs: >-
type=image,name=git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please,push-by-digest=true,name-canonical=true,push=true
build-args: |
VERSION=${{ steps.version.outputs.VERSION }}
BUILD_DATE=${{ steps.version.outputs.BUILD_DATE }}
COMMIT=${{ steps.version.outputs.COMMIT }}
- name: Export digest
run: |
mkdir -p /tmp/digests
digest="${{ steps.build.outputs.digest }}"
touch "/tmp/digests/${digest#sha256:}"
- name: Upload digest
uses: actions/upload-artifact@v7
with:
name: digests-linux-arm64
path: /tmp/digests/*
if-no-files-found: error
retention-days: 1
merge:
name: Merge Multi-Arch Manifest
build-and-push:
name: Build and Push Docker Image
runs-on: ubuntu-latest
needs: [build-amd64, build-arm64]
needs: test
steps:
- name: Download all digests
uses: actions/download-artifact@v8.0.1
with:
path: /tmp/digests
pattern: digests-*
merge-multiple: true
- name: Extract version
id: version
run: |
VERSION=${GITHUB_REF#refs/tags/v}
echo "VERSION=$VERSION" >> $GITHUB_OUTPUT
if echo "$VERSION" | grep -qE '^[0-9]+\.[0-9]+\.[0-9]+(-[a-zA-Z0-9.]+)?$'; then
echo "MAJOR=$(echo "$VERSION" | cut -d. -f1)" >> $GITHUB_OUTPUT
echo "MINOR=$(echo "$VERSION" | cut -d. -f2)" >> $GITHUB_OUTPUT
if echo "$VERSION" | grep -q '-'; then
echo "IS_PRERELEASE=true" >> $GITHUB_OUTPUT
else
echo "IS_PRERELEASE=false" >> $GITHUB_OUTPUT
fi
else
echo "Invalid version format: $VERSION"
exit 1
fi
- name: Checkout repository
uses: actions/checkout@v6
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
@@ -201,23 +49,64 @@ jobs:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Create and push manifest (release)
working-directory: /tmp/digests
if: steps.version.outputs.IS_PRERELEASE == 'false'
- name: Extract version and determine release type
id: version
run: |
docker buildx imagetools create \
-t git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:v${{ steps.version.outputs.VERSION }} \
-t git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:release \
-t git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:v${{ steps.version.outputs.MAJOR }}.${{ steps.version.outputs.MINOR }} \
-t git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:v${{ steps.version.outputs.MAJOR }} \
-t git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:latest \
$(printf 'git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please@sha256:%s ' *)
VERSION=${GITHUB_REF#refs/tags/v}
echo "VERSION=$VERSION" >> $GITHUB_OUTPUT
echo "BUILD_DATE=$(date -u +'%Y-%m-%dT%H:%M:%SZ')" >> $GITHUB_OUTPUT
echo "COMMIT=${{ github.sha }}" >> $GITHUB_OUTPUT
if echo "$VERSION" | grep -qE '^[0-9]+\.[0-9]+\.[0-9]+(-[a-zA-Z0-9.]+)?$'; then
MAJOR=$(echo "$VERSION" | cut -d. -f1)
MINOR=$(echo "$VERSION" | cut -d. -f2)
PATCH=$(echo "$VERSION" | cut -d. -f3 | cut -d- -f1)
echo "MAJOR=$MAJOR" >> $GITHUB_OUTPUT
echo "MINOR=$MINOR" >> $GITHUB_OUTPUT
echo "PATCH=$PATCH" >> $GITHUB_OUTPUT
if echo "$VERSION" | grep -q '-'; then
PRERELEASE_TAG=$(echo "$VERSION" | cut -d- -f2 | cut -d. -f1)
echo "IS_PRERELEASE=true" >> $GITHUB_OUTPUT
echo "PRERELEASE_TAG=$PRERELEASE_TAG" >> $GITHUB_OUTPUT
else
echo "IS_PRERELEASE=false" >> $GITHUB_OUTPUT
fi
else
echo "Invalid version format: $VERSION"
exit 1
fi
- name: Create and push manifest (pre-release)
working-directory: /tmp/digests
if: steps.version.outputs.IS_PRERELEASE == 'true'
run: |
docker buildx imagetools create \
-t git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:v${{ steps.version.outputs.VERSION }} \
-t git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:staging \
$(printf 'git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please@sha256:%s ' *)
- name: Build and push Docker image (release)
uses: docker/build-push-action@v6
with:
context: .
push: true
tags: |
git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:v${{ steps.version.outputs.VERSION }}
git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:release
git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:v${{ steps.version.outputs.MAJOR }}.${{ steps.version.outputs.MINOR }}
git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:v${{ steps.version.outputs.MAJOR }}
git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:latest
platforms: linux/amd64,linux/arm64
build-args: |
VERSION=${{ steps.version.outputs.VERSION }}
BUILD_DATE=${{ steps.version.outputs.BUILD_DATE }}
COMMIT=${{ steps.version.outputs.COMMIT }}
if: steps.version.outputs.IS_PRERELEASE == 'false'
- name: Build and push Docker image (pre-release)
uses: docker/build-push-action@v6
with:
context: .
push: true
tags: |
git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:v${{ steps.version.outputs.VERSION }}
git.fossy.my.id/${{ secrets.DOCKER_USERNAME }}/tunnel-please:staging
platforms: linux/amd64,linux/arm64
build-args: |
VERSION=${{ steps.version.outputs.VERSION }}
BUILD_DATE=${{ steps.version.outputs.BUILD_DATE }}
COMMIT=${{ steps.version.outputs.COMMIT }}
if: steps.version.outputs.IS_PRERELEASE == 'true'
+3 -4
View File
@@ -13,7 +13,6 @@ import (
"tunnel_pls/internal/session/slug"
"tunnel_pls/internal/types"
"tunnel_pls/internal/port"
"tunnel_pls/internal/registry"
proto "git.fossy.my.id/bagas/tunnel-please-grpc/gen"
@@ -885,16 +884,16 @@ func (m *mockLifecycle) Connection() ssh.Conn {
return args.Get(0).(ssh.Conn)
}
func (m *mockLifecycle) User() string { return m.Called().String(0) }
func (m *mockLifecycle) SetChannel(channel ssh.Channel) { m.Called(channel) }
func (m *mockLifecycle) SetChannel(channel ssh.Channel) error { return m.Called(channel).Error(0) }
func (m *mockLifecycle) SetStatus(status types.SessionStatus) { m.Called(status) }
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) PortRegistry() port.Port {
func (m *mockLifecycle) PortRegistry() lifecycle.PortRegistry {
args := m.Called()
if args.Get(0) == nil {
return nil
}
return args.Get(0).(port.Port)
return args.Get(0).(lifecycle.PortRegistry)
}
type mockEventServiceClient struct {
+3 -4
View File
@@ -4,7 +4,6 @@ import (
"sync"
"testing"
"time"
"tunnel_pls/internal/port"
"tunnel_pls/internal/session/forwarder"
"tunnel_pls/internal/session/interaction"
"tunnel_pls/internal/session/lifecycle"
@@ -78,15 +77,15 @@ func (ml *mockLifecycle) Connection() ssh.Conn {
return args.Get(0).(ssh.Conn)
}
func (ml *mockLifecycle) PortRegistry() port.Port {
func (ml *mockLifecycle) PortRegistry() lifecycle.PortRegistry {
args := ml.Called()
if args.Get(0) == nil {
return nil
}
return args.Get(0).(port.Port)
return args.Get(0).(lifecycle.PortRegistry)
}
func (ml *mockLifecycle) SetChannel(channel ssh.Channel) { ml.Called(channel) }
func (ml *mockLifecycle) SetChannel(channel ssh.Channel) error { return ml.Called(channel).Error(0) }
func (ml *mockLifecycle) SetStatus(status types.SessionStatus) { ml.Called(status) }
func (ml *mockLifecycle) IsActive() bool { return ml.Called().Bool(0) }
func (ml *mockLifecycle) StartedAt() time.Time { return ml.Called().Get(0).(time.Time) }
+16 -3
View File
@@ -28,6 +28,7 @@ type Forwarder interface {
Close() error
}
type forwarder struct {
mu sync.RWMutex
listener net.Listener
tunnelType types.TunnelType
forwardedPort uint16
@@ -60,7 +61,7 @@ func (f *forwarder) copyWithBuffer(dst io.Writer, src io.Reader) (written int64,
}
func (f *forwarder) OpenForwardedChannel(ctx context.Context, origin net.Addr) (ssh.Channel, <-chan *ssh.Request, error) {
payload := createForwardedTCPIPPayload(origin, f.forwardedPort)
payload := createForwardedTCPIPPayload(origin, f.ForwardedPort())
type channelResult struct {
channel ssh.Channel
reqs <-chan *ssh.Request
@@ -141,32 +142,44 @@ func (f *forwarder) HandleConnection(dst io.ReadWriter, src ssh.Channel) {
}
func (f *forwarder) SetType(tunnelType types.TunnelType) {
f.mu.Lock()
defer f.mu.Unlock()
f.tunnelType = tunnelType
}
func (f *forwarder) TunnelType() types.TunnelType {
f.mu.RLock()
defer f.mu.RUnlock()
return f.tunnelType
}
func (f *forwarder) ForwardedPort() uint16 {
f.mu.RLock()
defer f.mu.RUnlock()
return f.forwardedPort
}
func (f *forwarder) SetForwardedPort(port uint16) {
f.mu.Lock()
defer f.mu.Unlock()
f.forwardedPort = port
}
func (f *forwarder) SetListener(listener net.Listener) {
f.mu.Lock()
defer f.mu.Unlock()
f.listener = listener
}
func (f *forwarder) Listener() net.Listener {
f.mu.RLock()
defer f.mu.RUnlock()
return f.listener
}
func (f *forwarder) Close() error {
if f.Listener() != nil {
return f.listener.Close()
if listener := f.Listener(); listener != nil {
return listener.Close()
}
return nil
}
@@ -1922,10 +1922,6 @@ func TestInteraction_Start_ProtocolSelection(t *testing.T) {
time.Sleep(50 * time.Millisecond)
i := mockInteraction.(*interaction)
if i.program != nil {
assert.NotNil(t, i.program, "program should be initialized")
}
i.Stop()
mockConfig.AssertExpectations(t)
+78 -29
View File
@@ -2,6 +2,7 @@ package lifecycle
import (
"errors"
"fmt"
"io"
"net"
"sync"
@@ -9,8 +10,6 @@ import (
"tunnel_pls/internal/session/slug"
"tunnel_pls/internal/types"
portUtil "tunnel_pls/internal/port"
"golang.org/x/crypto/ssh"
)
@@ -24,6 +23,12 @@ type SessionRegistry interface {
Remove(key types.SessionKey)
}
type PortRegistry interface {
Unassigned() (uint16, bool)
Claim(port uint16) bool
SetStatus(port uint16, assigned bool) error
}
type lifecycle struct {
mu sync.Mutex
status types.SessionStatus
@@ -34,18 +39,18 @@ type lifecycle struct {
slug slug.Slug
startedAt time.Time
sessionRegistry SessionRegistry
portRegistry portUtil.Port
portRegistry PortRegistry
user string
}
func New(conn ssh.Conn, forwarder Forwarder, slugManager slug.Slug, port portUtil.Port, sessionRegistry SessionRegistry, user string) Lifecycle {
func New(conn ssh.Conn, forwarder Forwarder, slugManager slug.Slug, port PortRegistry, sessionRegistry SessionRegistry, user string) Lifecycle {
return &lifecycle{
status: types.SessionStatusINITIALIZING,
conn: conn,
channel: nil,
forwarder: forwarder,
slug: slugManager,
startedAt: time.Now(),
startedAt: time.Time{},
sessionRegistry: sessionRegistry,
portRegistry: port,
user: user,
@@ -55,16 +60,16 @@ func New(conn ssh.Conn, forwarder Forwarder, slugManager slug.Slug, port portUti
type Lifecycle interface {
Connection() ssh.Conn
Channel() ssh.Channel
PortRegistry() portUtil.Port
PortRegistry() PortRegistry
User() string
SetChannel(channel ssh.Channel)
SetChannel(channel ssh.Channel) error
SetStatus(status types.SessionStatus)
IsActive() bool
StartedAt() time.Time
Close() error
}
func (l *lifecycle) PortRegistry() portUtil.Port {
func (l *lifecycle) PortRegistry() PortRegistry {
return l.portRegistry
}
@@ -72,11 +77,25 @@ func (l *lifecycle) User() string {
return l.user
}
func (l *lifecycle) SetChannel(channel ssh.Channel) {
func (l *lifecycle) SetChannel(channel ssh.Channel) error {
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
return nil
}
func (l *lifecycle) Channel() ssh.Channel {
l.mu.Lock()
defer l.mu.Unlock()
return l.channel
}
@@ -87,7 +106,13 @@ func (l *lifecycle) Connection() ssh.Conn {
func (l *lifecycle) SetStatus(status types.SessionStatus) {
l.mu.Lock()
defer l.mu.Unlock()
if l.status == types.SessionStatusCLOSED {
return
}
l.status = status
if status == types.SessionStatusRUNNING && l.startedAt.IsZero() {
l.startedAt = time.Now()
}
}
func (l *lifecycle) IsActive() bool {
@@ -98,50 +123,74 @@ func (l *lifecycle) IsActive() bool {
func (l *lifecycle) Close() error {
l.mu.Lock()
defer l.mu.Unlock()
if l.status == types.SessionStatusCLOSED {
return l.closeErr
closeErr := l.closeErr
l.mu.Unlock()
return closeErr
}
l.status = types.SessionStatusCLOSED
channel := l.channel
conn := l.conn
l.mu.Unlock()
var errs []error
tunnelType := l.forwarder.TunnelType()
if l.channel != nil {
if err := l.channel.Close(); err != nil && !isClosedError(err) {
if channel != nil {
if err := channel.Close(); err != nil && !isClosedError(err) {
errs = append(errs, err)
}
}
if conn != nil {
if err := conn.Close(); err != nil && !isClosedError(err) {
errs = append(errs, err)
}
}
if l.conn != nil {
if err := l.conn.Close(); err != nil && !isClosedError(err) {
errs = append(errs, err)
}
l.cleanupRegistry()
if err := l.cleanupForwarder(); err != nil {
errs = append(errs, err)
}
clientSlug := l.slug.String()
closeErr := errors.Join(errs...)
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{
Id: clientSlug,
Type: tunnelType,
Id: slugStr,
Type: l.forwarder.TunnelType(),
}
l.sessionRegistry.Remove(key)
}
if tunnelType == types.TunnelTypeTCP {
errs = append(errs, l.PortRegistry().SetStatus(l.forwarder.ForwardedPort(), false))
errs = append(errs, l.forwarder.Close())
func (l *lifecycle) cleanupForwarder() error {
if l.forwarder.TunnelType() != types.TunnelTypeTCP {
return nil
}
l.closeErr = errors.Join(errs...)
return l.closeErr
var errs []error
errs = append(errs, l.portRegistry.SetStatus(l.forwarder.ForwardedPort(), false))
errs = append(errs, l.forwarder.Close())
return errors.Join(errs...)
}
func isClosedError(err error) bool {
if err == nil {
return false
}
return errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) || err.Error() == "EOF"
return errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed)
}
func (l *lifecycle) StartedAt() time.Time {
l.mu.Lock()
defer l.mu.Unlock()
return l.startedAt
}
+125 -4
View File
@@ -5,6 +5,7 @@ import (
"errors"
"io"
"net"
"sync"
"testing"
"tunnel_pls/internal/types"
@@ -177,8 +178,14 @@ func TestLifecycle_SetChannel(t *testing.T) {
mockSSHChannel := &MockSSHChannel{}
mockLifecycle.SetChannel(mockSSHChannel)
err := 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())
}
@@ -276,14 +283,15 @@ func TestLifecycle_Close(t *testing.T) {
mockLifecycle := New(mockSSHConn, mockForwarder, mockSlug, mockPort, mockSessionRegistry, "mas-fuad")
mockLifecycle.SetStatus(types.SessionStatusRUNNING)
mockLifecycle.SetChannel(mockSSHChannel)
err := mockLifecycle.SetChannel(mockSSHChannel)
assert.NoError(t, err)
if tt.alreadyClosed {
err := mockLifecycle.Close()
err = mockLifecycle.Close()
assert.NoError(t, err)
}
err := mockLifecycle.Close()
err = mockLifecycle.Close()
if tt.expectErr {
assert.Error(t, err)
@@ -301,3 +309,116 @@ 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")
}
+19 -19
View File
@@ -158,13 +158,15 @@ func (s *session) setupInteractiveMode(channel ssh.NewChannel) error {
}
go func() {
err = s.HandleGlobalRequest(reqs)
err := s.HandleGlobalRequest(reqs)
if err != nil {
log.Printf("global request handler error: %v", err)
}
}()
s.lifecycle.SetChannel(ch)
if err = s.lifecycle.SetChannel(ch); err != nil {
return err
}
s.interaction.SetChannel(ch)
s.interaction.SetMode(types.InteractiveModeINTERACTIVE)
@@ -198,23 +200,22 @@ func (s *session) waitForSessionEnd() error {
}
func (s *session) waitForTCPIPForward() *ssh.Request {
select {
case req, ok := <-s.initialReq:
if !ok {
log.Println("Forwarding request channel closed")
for {
select {
case req, ok := <-s.initialReq:
if !ok {
log.Println("Forwarding request channel closed")
return nil
}
if req.Type == "tcpip-forward" {
return req
}
log.Printf("Ignoring unexpected global request: %s", req.Type)
_ = req.Reply(false, nil)
case <-time.After(500 * time.Millisecond):
log.Println("No tcpip-forward request received within timeout")
return nil
}
if req.Type == "tcpip-forward" {
return req
}
if err := req.Reply(false, nil); err != nil {
log.Printf("Failed to reply to request: %v", err)
}
log.Printf("Expected tcpip-forward request, got: %s", req.Type)
return nil
case <-time.After(500 * time.Millisecond):
log.Println("No forwarding request received")
return nil
}
}
@@ -377,8 +378,7 @@ func (s *session) HandleTCPForward(req *ssh.Request, addr string, portToBind uin
}
go func() {
err = tcpServer.Serve(listener)
if err != nil {
if err := tcpServer.Serve(listener); err != nil {
log.Printf("Failed serving tcp server: %s\n", err)
}
}()
+56 -1
View File
@@ -807,7 +807,7 @@ func (m *mockNewChanFail) Accept() (ssh.Channel, <-chan *ssh.Request, error) {
}
func TestWaitForTCPIPForward_EdgeCases(t *testing.T) {
t.Run("Wrong Request Type", func(t *testing.T) {
t.Run("Wrong Request Type Then Timeout", func(t *testing.T) {
_, sReqs, _, cConn, cleanup := setupSSH(t)
defer cleanup()
@@ -817,10 +817,65 @@ func TestWaitForTCPIPForward_EdgeCases(t *testing.T) {
_, _, _ = cConn.SendRequest("not-tcpip-forward", true, nil)
}()
start := time.Now()
req := s.waitForTCPIPForward()
elapsed := time.Since(start)
if req != nil {
t.Error("expected nil request")
}
if elapsed < 400*time.Millisecond {
t.Errorf("expected timeout ~500ms, got %v", elapsed)
}
})
t.Run("Multiple Non-Forward Requests Then Success", func(t *testing.T) {
_, sReqs, _, cConn, cleanup := setupSSH(t)
defer cleanup()
s := &session{initialReq: sReqs}
go func() {
time.Sleep(100 * time.Millisecond)
_, _, _ = cConn.SendRequest("keepalive@openssh.com", false, nil)
time.Sleep(100 * time.Millisecond)
_, _, _ = cConn.SendRequest("hostkeys-00@openssh.com", false, nil)
time.Sleep(100 * time.Millisecond)
_, _, _ = cConn.SendRequest("tcpip-forward", true, nil)
}()
req := s.waitForTCPIPForward()
if req == nil {
t.Error("expected tcpip-forward request, got nil")
}
if req != nil && req.Type != "tcpip-forward" {
t.Errorf("expected tcpip-forward, got %s", req.Type)
}
})
t.Run("Timeout After Non-Forward Requests", func(t *testing.T) {
_, sReqs, _, cConn, cleanup := setupSSH(t)
defer cleanup()
s := &session{initialReq: sReqs}
go func() {
time.Sleep(100 * time.Millisecond)
_, _, _ = cConn.SendRequest("keepalive@openssh.com", false, nil)
time.Sleep(100 * time.Millisecond)
_, _, _ = cConn.SendRequest("hostkeys-00@openssh.com", false, nil)
}()
start := time.Now()
req := s.waitForTCPIPForward()
elapsed := time.Since(start)
if req != nil {
t.Error("expected nil request after timeout")
}
if elapsed < 400*time.Millisecond {
t.Errorf("expected timeout ~500ms after last request, got %v", elapsed)
}
})
t.Run("Channel Closed", func(t *testing.T) {
+7
View File
@@ -1,11 +1,14 @@
package slug
import "sync"
type Slug interface {
String() string
Set(slug string)
}
type slug struct {
mu sync.RWMutex
slug string
}
@@ -16,9 +19,13 @@ func New() Slug {
}
func (s *slug) String() string {
s.mu.RLock()
defer s.mu.RUnlock()
return s.slug
}
func (s *slug) Set(slug string) {
s.mu.Lock()
defer s.mu.Unlock()
s.slug = slug
}
+1 -2
View File
@@ -55,8 +55,7 @@ func TestHTTPServer_Serve(t *testing.T) {
go func() {
time.Sleep(100 * time.Millisecond)
err = listener.Close()
assert.NoError(t, err)
_ = listener.Close()
}()
err = srv.Serve(listener)
+1 -2
View File
@@ -63,8 +63,7 @@ func TestHTTPSServer_Serve(t *testing.T) {
go func() {
time.Sleep(100 * time.Millisecond)
err = listener.Close()
assert.NoError(t, err)
_ = listener.Close()
}()
err = srv.Serve(listener)
+1 -2
View File
@@ -45,8 +45,7 @@ func TestTCPServer_Serve(t *testing.T) {
go func() {
time.Sleep(100 * time.Millisecond)
err = listener.Close()
assert.NoError(t, err)
_ = listener.Close()
}()
err = srv.Serve(listener)
+3 -6
View File
@@ -2,8 +2,6 @@
"extends": [
"config:recommended"
],
"prConcurrentLimit": 1,
"prHourlyLimit": 1,
"packageRules": [
{
"matchUpdateTypes": [
@@ -12,11 +10,10 @@
"pin",
"digest"
],
"groupName": "all-dependencies",
"automerge": true,
"matchPackageNames": [
"*"
"baseBranchPatterns": [
"staging"
]
}
]
}
}