diff --git a/internal/session/query.go b/internal/session/query.go index 7a10af70f..97a8c57f8 100644 --- a/internal/session/query.go +++ b/internal/session/query.go @@ -388,6 +388,17 @@ where and ss.end_time is null %s ; +` + deleteTerminated = ` +delete from session +using session_state +where + session.public_id = session_state.session_id +and + session_state.state = 'terminated' +and + session_state.start_time < wt_sub_seconds_from_now(@threshold_seconds) +; ` ) diff --git a/internal/session/repository_session.go b/internal/session/repository_session.go index d16edfa21..a6acb6fce 100644 --- a/internal/session/repository_session.go +++ b/internal/session/repository_session.go @@ -7,6 +7,7 @@ import ( "database/sql" "fmt" "strings" + "time" "github.com/hashicorp/boundary/internal/boundary" "github.com/hashicorp/boundary/internal/db" @@ -700,6 +701,20 @@ func (r *Repository) checkIfNoLongerActive(ctx context.Context, reportedSessions return notActive, nil } +func (r *Repository) deleteSessionsTerminatedBefore(ctx context.Context, threshold time.Duration) (int, error) { + const op = "session.(Repository).deleteTerminated" + + args := []any{ + sql.Named("threshold_seconds", threshold.Seconds()), + } + + c, err := r.writer.Exec(ctx, deleteTerminated, args) + if err != nil { + return 0, errors.Wrap(ctx, err, op, errors.WithMsg("error deleting terminated sessions")) + } + return c, nil +} + func fetchStates(ctx context.Context, r db.Reader, sessionId string, opt ...db.Option) ([]*State, error) { const op = "session.fetchStates" var states []*State diff --git a/internal/session/repository_session_test.go b/internal/session/repository_session_test.go index 408633743..891354b6b 100644 --- a/internal/session/repository_session_test.go +++ b/internal/session/repository_session_test.go @@ -1503,3 +1503,91 @@ func TestRepository_deleteTargetFKey(t *testing.T) { }) } } + +func TestRepository_deleteTerminated(t *testing.T) { + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + iamRepo := iam.TestRepo(t, conn, wrapper) + kms := kms.TestKms(t, conn, wrapper) + repo, err := NewRepository(rw, rw, kms) + composedOf := TestSessionParams(t, conn, wrapper, iamRepo) + + cases := []struct { + sessionCount int + terminateCount int + threshold time.Duration + expected int + }{ + { + 0, + 0, + time.Nanosecond, + 0, + }, + { + 1, + 1, + time.Nanosecond, + 1, + }, + { + 1, + 1, + time.Hour, + 0, + }, + { + 10, + 10, + time.Nanosecond, + 10, + }, + { + 10, + 4, + time.Nanosecond, + 4, + }, + { + 10, + 0, + time.Nanosecond, + 0, + }, + { + 10, + 10, + time.Hour, + 0, + }, + } + + for _, tc := range cases { + t.Run(fmt.Sprintf("%d_%d_%s", tc.sessionCount, tc.terminateCount, tc.threshold), func(t *testing.T) { + t.Cleanup(func() { + sdb, err := conn.SqlDB(ctx) + require.NoError(t, err) + _, err = sdb.Exec(`delete from session;`) + require.NoError(t, err) + }) + + for i := 0; i < tc.sessionCount; i++ { + s := TestSession(t, conn, wrapper, composedOf) + if i < tc.terminateCount { + _, err = repo.CancelSession(ctx, s.PublicId, s.Version) + require.NoError(t, err) + } + + } + c, err := repo.TerminateCompletedSessions(ctx) + require.NoError(t, err) + assert.Equal(t, tc.terminateCount, c) + + c, err = repo.deleteSessionsTerminatedBefore(ctx, tc.threshold) + require.NoError(t, err) + assert.Equal(t, tc.expected, c) + }) + } +}