feat: Add buffering to provisioner job logs (#4918)

* feat: Add bufferring to provisioner job logs

This should improve overall build performance, and especially under load.

It removes the old `id` column on the `provisioner_job_logs` table
and replaces it with an auto-incrementing big integer to preserve order.

Funny enough, we never had to care about order before because inserts
would at minimum be 1ms different. Now they aren't, so the order needs
to be preserved.

* Fix log bufferring

* Fix frontend log streaming

* Fix JS test

coderd/database/databasefake/databasefake.go

@@ -2052,17 +2052,14 @@ func (q *fakeQuerier) GetProvisionerLogsByIDBetween(_ context.Context, arg datab
 		if jobLog.JobID != arg.JobID {
 			continue
 		}
-		if !arg.CreatedBefore.IsZero() && jobLog.CreatedAt.After(arg.CreatedBefore) {
+		if arg.CreatedBefore != 0 && jobLog.ID > arg.CreatedBefore {
 			continue
 		}
-		if !arg.CreatedAfter.IsZero() && jobLog.CreatedAt.Before(arg.CreatedAfter) {
+		if arg.CreatedAfter != 0 && jobLog.ID < arg.CreatedAfter {
 			continue
 		}
 		logs = append(logs, jobLog)
 	}
-	if len(logs) == 0 {
-		return nil, sql.ErrNoRows
-	}
 	return logs, nil
 }
 
@@ -2212,10 +2209,15 @@ func (q *fakeQuerier) InsertProvisionerJobLogs(_ context.Context, arg database.I
 	defer q.mutex.Unlock()
 
 	logs := make([]database.ProvisionerJobLog, 0)
+	id := int64(1)
+	if len(q.provisionerJobLogs) > 0 {
+		id = q.provisionerJobLogs[len(q.provisionerJobLogs)-1].ID
+	}
 	for index, output := range arg.Output {
+		id++
 		logs = append(logs, database.ProvisionerJobLog{
+			ID:        id,
 			JobID:     arg.JobID,
-			ID:        arg.ID[index],
 			CreatedAt: arg.CreatedAt[index],
 			Source:    arg.Source[index],
 			Level:     arg.Level[index],

coderd/database/dump.sql

@@ -272,15 +272,24 @@ CREATE TABLE provisioner_daemons (
 );
 
 CREATE TABLE provisioner_job_logs (
-    id uuid NOT NULL,
     job_id uuid NOT NULL,
     created_at timestamp with time zone NOT NULL,
     source log_source NOT NULL,
     level log_level NOT NULL,
     stage character varying(128) NOT NULL,
-    output character varying(1024) NOT NULL
+    output character varying(1024) NOT NULL,
+    id bigint NOT NULL
 );
 
+CREATE SEQUENCE provisioner_job_logs_id_seq
+    START WITH 1
+    INCREMENT BY 1
+    NO MINVALUE
+    NO MAXVALUE
+    CACHE 1;
+
+ALTER SEQUENCE provisioner_job_logs_id_seq OWNED BY provisioner_job_logs.id;
+
 CREATE TABLE provisioner_jobs (
     id uuid NOT NULL,
     created_at timestamp with time zone NOT NULL,
@@ -463,6 +472,8 @@ CREATE TABLE workspaces (
 
 ALTER TABLE ONLY licenses ALTER COLUMN id SET DEFAULT nextval('licenses_id_seq'::regclass);
 
+ALTER TABLE ONLY provisioner_job_logs ALTER COLUMN id SET DEFAULT nextval('provisioner_job_logs_id_seq'::regclass);
+
 ALTER TABLE ONLY agent_stats
     ADD CONSTRAINT agent_stats_pkey PRIMARY KEY (id);
 

coderd/database/migrations/000071_provisioner_log_lines.down.sql

@@ -0,0 +1,3 @@
+ALTER TABLE provisioner_job_logs DROP COLUMN id;
+
+ALTER TABLE provisioner_job_logs ADD COLUMN id uuid NOT NULL DEFAULT gen_random_uuid();

coderd/database/migrations/000071_provisioner_log_lines.up.sql

@@ -0,0 +1,3 @@
+ALTER TABLE provisioner_job_logs DROP COLUMN id;
+
+ALTER TABLE provisioner_job_logs ADD COLUMN id BIGSERIAL PRIMARY KEY; 

coderd/database/models.go

@@ -545,13 +545,13 @@ type ProvisionerJob struct {
 }
 
 type ProvisionerJobLog struct {
-	ID        uuid.UUID `db:"id" json:"id"`
 	JobID     uuid.UUID `db:"job_id" json:"job_id"`
 	CreatedAt time.Time `db:"created_at" json:"created_at"`
 	Source    LogSource `db:"source" json:"source"`
 	Level     LogLevel  `db:"level" json:"level"`
 	Stage     string    `db:"stage" json:"stage"`
 	Output    string    `db:"output" json:"output"`
+	ID        int64     `db:"id" json:"id"`
 }
 
 type Replica struct {

coderd/database/queries.sql.go

@@ -2345,23 +2345,21 @@ func (q *sqlQuerier) UpdateProvisionerDaemonByID(ctx context.Context, arg Update
 
 const getProvisionerLogsByIDBetween = `-- name: GetProvisionerLogsByIDBetween :many
 SELECT
-	id, job_id, created_at, source, level, stage, output
+	job_id, created_at, source, level, stage, output, id
 FROM
 	provisioner_job_logs
 WHERE
 	job_id = $1
 	AND (
-		created_at >= $2
-		OR created_at <= $3
-	)
-ORDER BY
-	created_at DESC
+		id > $2
+		OR id < $3
+	) ORDER BY id
 `
 
 type GetProvisionerLogsByIDBetweenParams struct {
 	JobID         uuid.UUID `db:"job_id" json:"job_id"`
-	CreatedAfter  time.Time `db:"created_after" json:"created_after"`
-	CreatedBefore time.Time `db:"created_before" json:"created_before"`
+	CreatedAfter  int64     `db:"created_after" json:"created_after"`
+	CreatedBefore int64     `db:"created_before" json:"created_before"`
 }
 
 func (q *sqlQuerier) GetProvisionerLogsByIDBetween(ctx context.Context, arg GetProvisionerLogsByIDBetweenParams) ([]ProvisionerJobLog, error) {
@@ -2374,13 +2372,13 @@ func (q *sqlQuerier) GetProvisionerLogsByIDBetween(ctx context.Context, arg GetP
 	for rows.Next() {
 		var i ProvisionerJobLog
 		if err := rows.Scan(
-			&i.ID,
 			&i.JobID,
 			&i.CreatedAt,
 			&i.Source,
 			&i.Level,
 			&i.Stage,
 			&i.Output,
+			&i.ID,
 		); err != nil {
 			return nil, err
 		}
@@ -2399,17 +2397,15 @@ const insertProvisionerJobLogs = `-- name: InsertProvisionerJobLogs :many
 INSERT INTO
 	provisioner_job_logs
 SELECT
-	unnest($1 :: uuid [ ]) AS id,
-	$2 :: uuid AS job_id,
-	unnest($3 :: timestamptz [ ]) AS created_at,
-	unnest($4 :: log_source [ ]) AS source,
-	unnest($5 :: log_level [ ]) AS LEVEL,
-	unnest($6 :: VARCHAR(128) [ ]) AS stage,
-	unnest($7 :: VARCHAR(1024) [ ]) AS output RETURNING id, job_id, created_at, source, level, stage, output
+	$1 :: uuid AS job_id,
+	unnest($2 :: timestamptz [ ]) AS created_at,
+	unnest($3 :: log_source [ ]) AS source,
+	unnest($4 :: log_level [ ]) AS LEVEL,
+	unnest($5 :: VARCHAR(128) [ ]) AS stage,
+	unnest($6 :: VARCHAR(1024) [ ]) AS output RETURNING job_id, created_at, source, level, stage, output, id
 `
 
 type InsertProvisionerJobLogsParams struct {
-	ID        []uuid.UUID `db:"id" json:"id"`
 	JobID     uuid.UUID   `db:"job_id" json:"job_id"`
 	CreatedAt []time.Time `db:"created_at" json:"created_at"`
 	Source    []LogSource `db:"source" json:"source"`
@@ -2420,7 +2416,6 @@ type InsertProvisionerJobLogsParams struct {
 
 func (q *sqlQuerier) InsertProvisionerJobLogs(ctx context.Context, arg InsertProvisionerJobLogsParams) ([]ProvisionerJobLog, error) {
 	rows, err := q.db.QueryContext(ctx, insertProvisionerJobLogs,
-		pq.Array(arg.ID),
 		arg.JobID,
 		pq.Array(arg.CreatedAt),
 		pq.Array(arg.Source),
@@ -2436,13 +2431,13 @@ func (q *sqlQuerier) InsertProvisionerJobLogs(ctx context.Context, arg InsertPro
 	for rows.Next() {
 		var i ProvisionerJobLog
 		if err := rows.Scan(
-			&i.ID,
 			&i.JobID,
 			&i.CreatedAt,
 			&i.Source,
 			&i.Level,
 			&i.Stage,
 			&i.Output,
+			&i.ID,
 		); err != nil {
 			return nil, err
 		}

coderd/database/queries/provisionerjoblogs.sql

@@ -6,17 +6,14 @@ FROM
 WHERE
 	job_id = @job_id
 	AND (
-		created_at >= @created_after
-		OR created_at <= @created_before
-	)
-ORDER BY
-	created_at DESC;
+		id > @created_after
+		OR id < @created_before
+	) ORDER BY id;
 
 -- name: InsertProvisionerJobLogs :many
 INSERT INTO
 	provisioner_job_logs
 SELECT
-	unnest(@id :: uuid [ ]) AS id,
 	@job_id :: uuid AS job_id,
 	unnest(@created_at :: timestamptz [ ]) AS created_at,
 	unnest(@source :: log_source [ ]) AS source,

coderd/provisionerdaemons.go

@@ -368,7 +368,6 @@ func (server *provisionerdServer) UpdateJob(ctx context.Context, request *proto.
 			if err != nil {
 				return nil, xerrors.Errorf("convert log source: %w", err)
 			}
-			insertParams.ID = append(insertParams.ID, uuid.New())
 			insertParams.CreatedAt = append(insertParams.CreatedAt, time.UnixMilli(log.CreatedAt))
 			insertParams.Level = append(insertParams.Level, logLevel)
 			insertParams.Stage = append(insertParams.Stage, log.Stage)
@@ -384,10 +383,15 @@ func (server *provisionerdServer) UpdateJob(ctx context.Context, request *proto.
 			server.Logger.Error(ctx, "failed to insert job logs", slog.F("job_id", parsedID), slog.Error(err))
 			return nil, xerrors.Errorf("insert job logs: %w", err)
 		}
+		// Publish by the lowest log ID inserted so the
+		// log stream will fetch everything from that point.
+		lowestID := logs[0].ID
 		server.Logger.Debug(ctx, "inserted job logs", slog.F("job_id", parsedID))
-		data, err := json.Marshal(provisionerJobLogsMessage{Logs: logs})
+		data, err := json.Marshal(provisionerJobLogsMessage{
+			CreatedAfter: lowestID,
+		})
 		if err != nil {
-			return nil, xerrors.Errorf("marshal job log: %w", err)
+			return nil, xerrors.Errorf("marshal: %w", err)
 		}
 		err = server.Pubsub.Publish(provisionerJobLogsChannel(parsedID), data)
 		if err != nil {

coderd/provisionerjobs.go

@@ -24,8 +24,8 @@ import (
 // Returns provisioner logs based on query parameters.
 // The intended usage for a client to stream all logs (with JS API):
 // const timestamp = new Date().getTime();
-// 1. GET /logs?before=<timestamp>
-// 2. GET /logs?after=<timestamp>&follow
+// 1. GET /logs?before=<id>
+// 2. GET /logs?after=<id>&follow
 // The combination of these responses should provide all current logs
 // to the consumer, and future logs are streamed in the follow request.
 func (api *API) provisionerJobLogs(rw http.ResponseWriter, r *http.Request, job database.ProvisionerJob) {
@@ -74,10 +74,11 @@ func (api *API) provisionerJobLogs(rw http.ResponseWriter, r *http.Request, job
 		}
 	}
 
-	var after time.Time
+	var after int64
 	// Only fetch logs created after the time provided.
 	if afterRaw != "" {
-		afterMS, err := strconv.ParseInt(afterRaw, 10, 64)
+		var err error
+		after, err = strconv.ParseInt(afterRaw, 10, 64)
 		if err != nil {
 			httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
 				Message: "Query param \"after\" must be an integer.",
@@ -87,16 +88,12 @@ func (api *API) provisionerJobLogs(rw http.ResponseWriter, r *http.Request, job
 			})
 			return
 		}
-		after = time.UnixMilli(afterMS)
-	} else {
-		if follow {
-			after = database.Now()
-		}
 	}
-	var before time.Time
+	var before int64
 	// Only fetch logs created before the time provided.
 	if beforeRaw != "" {
-		beforeMS, err := strconv.ParseInt(beforeRaw, 10, 64)
+		var err error
+		before, err = strconv.ParseInt(beforeRaw, 10, 64)
 		if err != nil {
 			httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
 				Message: "Query param \"before\" must be an integer.",
@@ -106,12 +103,6 @@ func (api *API) provisionerJobLogs(rw http.ResponseWriter, r *http.Request, job
 			})
 			return
 		}
-		before = time.UnixMilli(beforeMS)
-	} else {
-		// If we're following, we don't want logs before a timestamp!
-		if !follow {
-			before = database.Now()
-		}
 	}
 
 	logs, err := api.Database.GetProvisionerLogsByIDBetween(ctx, database.GetProvisionerLogsByIDBetweenParams{
@@ -156,7 +147,7 @@ func (api *API) provisionerJobLogs(rw http.ResponseWriter, r *http.Request, job
 	ctx, wsNetConn := websocketNetConn(ctx, conn, websocket.MessageText)
 	defer wsNetConn.Close() // Also closes conn.
 
-	logIdsDone := make(map[uuid.UUID]bool)
+	logIdsDone := make(map[int64]bool)
 
 	// The Go stdlib JSON encoder appends a newline character after message write.
 	encoder := json.NewEncoder(wsNetConn)
@@ -370,8 +361,8 @@ func provisionerJobLogsChannel(jobID uuid.UUID) string {
 
 // provisionerJobLogsMessage is the message type published on the provisionerJobLogsChannel() channel
 type provisionerJobLogsMessage struct {
-	EndOfLogs bool                         `json:"end_of_logs,omitempty"`
-	Logs      []database.ProvisionerJobLog `json:"logs,omitempty"`
+	CreatedAfter int64 `json:"created_after"`
+	EndOfLogs    bool  `json:"end_of_logs,omitempty"`
 }
 
 func (api *API) followLogs(jobID uuid.UUID) (<-chan database.ProvisionerJobLog, func(), error) {
@@ -389,23 +380,32 @@ func (api *API) followLogs(jobID uuid.UUID) (<-chan database.ProvisionerJobLog,
 				return
 			default:
 			}
-
 			jlMsg := provisionerJobLogsMessage{}
 			err := json.Unmarshal(message, &jlMsg)
 			if err != nil {
 				logger.Warn(ctx, "invalid provisioner job log on channel", slog.Error(err))
 				return
 			}
+			if jlMsg.CreatedAfter != 0 {
+				logs, err := api.Database.GetProvisionerLogsByIDBetween(ctx, database.GetProvisionerLogsByIDBetweenParams{
+					JobID:        jobID,
+					CreatedAfter: jlMsg.CreatedAfter,
+				})
+				if err != nil {
+					logger.Warn(ctx, "get provisioner logs", slog.Error(err))
+					return
+				}
 
-			for _, log := range jlMsg.Logs {
-				select {
-				case bufferedLogs <- log:
-					logger.Debug(ctx, "subscribe buffered log", slog.F("stage", log.Stage))
-				default:
-					// If this overflows users could miss logs streaming. This can happen
-					// we get a lot of logs and consumer isn't keeping up.  We don't want to block the pubsub,
-					// so just drop them.
-					logger.Warn(ctx, "provisioner job log overflowing channel")
+				for _, log := range logs {
+					select {
+					case bufferedLogs <- log:
+						logger.Debug(ctx, "subscribe buffered log", slog.F("stage", log.Stage))
+					default:
+						// If this overflows users could miss logs streaming. This can happen
+						// we get a lot of logs and consumer isn't keeping up.  We don't want to block the pubsub,
+						// so just drop them.
+						logger.Warn(ctx, "provisioner job log overflowing channel")
+					}
 				}
 			}
 			if jlMsg.EndOfLogs {

coderd/provisionerjobs_internal_test.go

@@ -1,160 +1,15 @@
 package coderd
 
 import (
-	"context"
-	"crypto/sha256"
 	"database/sql"
-	"encoding/json"
-	"net/http/httptest"
-	"net/url"
-	"sync"
 	"testing"
-	"time"
 
-	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 
-	"cdr.dev/slog"
-	"cdr.dev/slog/sloggers/slogtest"
 	"github.com/coder/coder/coderd/database"
-	"github.com/coder/coder/coderd/database/databasefake"
-	"github.com/coder/coder/coderd/rbac"
 	"github.com/coder/coder/codersdk"
-	"github.com/coder/coder/testutil"
 )
 
-func TestProvisionerJobLogs_Unit(t *testing.T) {
-	t.Parallel()
-
-	t.Run("QueryPubSubDupes", func(t *testing.T) {
-		t.Parallel()
-		logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
-		// mDB := mocks.NewStore(t)
-		fDB := databasefake.New()
-		fPubsub := &fakePubSub{t: t, cond: sync.NewCond(&sync.Mutex{})}
-		opts := Options{
-			Logger:   logger,
-			Database: fDB,
-			Pubsub:   fPubsub,
-		}
-		api := New(&opts)
-		defer api.Close()
-
-		server := httptest.NewServer(api.RootHandler)
-		defer server.Close()
-		userID := uuid.New()
-		keyID, keySecret, err := generateAPIKeyIDSecret()
-		require.NoError(t, err)
-		hashed := sha256.Sum256([]byte(keySecret))
-
-		u, err := url.Parse(server.URL)
-		require.NoError(t, err)
-		client := codersdk.Client{
-			HTTPClient:   server.Client(),
-			SessionToken: keyID + "-" + keySecret,
-			URL:          u,
-		}
-
-		buildID := uuid.New()
-		workspaceID := uuid.New()
-		jobID := uuid.New()
-
-		expectedLogs := []database.ProvisionerJobLog{
-			{ID: uuid.New(), JobID: jobID, Stage: "Stage0"},
-			{ID: uuid.New(), JobID: jobID, Stage: "Stage1"},
-			{ID: uuid.New(), JobID: jobID, Stage: "Stage2"},
-			{ID: uuid.New(), JobID: jobID, Stage: "Stage3"},
-		}
-
-		ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
-		defer cancel()
-
-		// wow there are a lot of DB rows we touch...
-		_, err = fDB.InsertAPIKey(ctx, database.InsertAPIKeyParams{
-			ID:           keyID,
-			HashedSecret: hashed[:],
-			UserID:       userID,
-			ExpiresAt:    time.Now().Add(5 * time.Hour),
-			LoginType:    database.LoginTypePassword,
-			Scope:        database.APIKeyScopeAll,
-		})
-		require.NoError(t, err)
-		_, err = fDB.InsertUser(ctx, database.InsertUserParams{
-			ID:        userID,
-			RBACRoles: []string{rbac.RoleOwner()},
-		})
-		require.NoError(t, err)
-		_, err = fDB.InsertWorkspaceBuild(ctx, database.InsertWorkspaceBuildParams{
-			ID:          buildID,
-			WorkspaceID: workspaceID,
-			JobID:       jobID,
-		})
-		require.NoError(t, err)
-		_, err = fDB.InsertWorkspace(ctx, database.InsertWorkspaceParams{
-			ID: workspaceID,
-		})
-		require.NoError(t, err)
-		_, err = fDB.InsertProvisionerJob(ctx, database.InsertProvisionerJobParams{
-			ID: jobID,
-		})
-		require.NoError(t, err)
-		for _, l := range expectedLogs[:2] {
-			_, err := fDB.InsertProvisionerJobLogs(ctx, database.InsertProvisionerJobLogsParams{
-				ID:    []uuid.UUID{l.ID},
-				JobID: jobID,
-				Stage: []string{l.Stage},
-			})
-			require.NoError(t, err)
-		}
-
-		logs, closer, err := client.WorkspaceBuildLogsAfter(ctx, buildID, time.Now())
-		require.NoError(t, err)
-		defer closer.Close()
-
-		// when the endpoint calls subscribe, we get the listener here.
-		fPubsub.cond.L.Lock()
-		for fPubsub.listener == nil {
-			fPubsub.cond.Wait()
-		}
-
-		// endpoint should now be listening
-		assert.False(t, fPubsub.canceled)
-		assert.False(t, fPubsub.closed)
-
-		// send all the logs in two batches, duplicating what we already returned on the DB query.
-		msg := provisionerJobLogsMessage{}
-		msg.Logs = expectedLogs[:2]
-		data, err := json.Marshal(msg)
-		require.NoError(t, err)
-		fPubsub.listener(ctx, data)
-		msg.Logs = expectedLogs[2:]
-		data, err = json.Marshal(msg)
-		require.NoError(t, err)
-		fPubsub.listener(ctx, data)
-
-		// send end of logs
-		msg.Logs = nil
-		msg.EndOfLogs = true
-		data, err = json.Marshal(msg)
-		require.NoError(t, err)
-		fPubsub.listener(ctx, data)
-
-		var stages []string
-		for l := range logs {
-			logger.Info(ctx, "got log",
-				slog.F("id", l.ID),
-				slog.F("stage", l.Stage))
-			stages = append(stages, l.Stage)
-		}
-		assert.Equal(t, []string{"Stage0", "Stage1", "Stage2", "Stage3"}, stages)
-		for !fPubsub.canceled {
-			fPubsub.cond.Wait()
-		}
-		assert.False(t, fPubsub.closed)
-	})
-}
-
 func TestConvertProvisionerJob_Unit(t *testing.T) {
 	t.Parallel()
 	validNullTimeMock := sql.NullTime{
@@ -260,39 +115,3 @@ func TestConvertProvisionerJob_Unit(t *testing.T) {
 		})
 	}
 }
-
-type fakePubSub struct {
-	t        *testing.T
-	cond     *sync.Cond
-	listener database.Listener
-	canceled bool
-	closed   bool
-}
-
-func (f *fakePubSub) Subscribe(_ string, listener database.Listener) (cancel func(), err error) {
-	f.cond.L.Lock()
-	defer f.cond.L.Unlock()
-	f.listener = listener
-	f.cond.Signal()
-	return f.cancel, nil
-}
-
-func (f *fakePubSub) Publish(_ string, _ []byte) error {
-	f.t.Fail()
-	return nil
-}
-
-func (f *fakePubSub) Close() error {
-	f.cond.L.Lock()
-	defer f.cond.L.Unlock()
-	f.closed = true
-	f.cond.Signal()
-	return nil
-}
-
-func (f *fakePubSub) cancel() {
-	f.cond.L.Lock()
-	defer f.cond.L.Unlock()
-	f.canceled = true
-	f.cond.Signal()
-}

coderd/provisionerjobs_test.go

@@ -3,12 +3,10 @@ package coderd_test
 import (
 	"context"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/require"
 
 	"github.com/coder/coder/coderd/coderdtest"
-	"github.com/coder/coder/coderd/database"
 	"github.com/coder/coder/provisioner/echo"
 	"github.com/coder/coder/provisionersdk/proto"
 	"github.com/coder/coder/testutil"
@@ -38,13 +36,12 @@ func TestProvisionerJobLogs(t *testing.T) {
 		template := coderdtest.CreateTemplate(t, client, user.OrganizationID, version.ID)
 		coderdtest.AwaitTemplateVersionJob(t, client, version.ID)
 		workspace := coderdtest.CreateWorkspace(t, client, user.OrganizationID, template.ID)
-		before := time.Now().UTC()
 		coderdtest.AwaitWorkspaceBuildJob(t, client, workspace.LatestBuild.ID)
 
 		ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
 		defer cancel()
 
-		logs, closer, err := client.WorkspaceBuildLogsAfter(ctx, workspace.LatestBuild.ID, before)
+		logs, closer, err := client.WorkspaceBuildLogsAfter(ctx, workspace.LatestBuild.ID, 0)
 		require.NoError(t, err)
 		defer closer.Close()
 		for {
@@ -78,12 +75,11 @@ func TestProvisionerJobLogs(t *testing.T) {
 		template := coderdtest.CreateTemplate(t, client, user.OrganizationID, version.ID)
 		coderdtest.AwaitTemplateVersionJob(t, client, version.ID)
 		workspace := coderdtest.CreateWorkspace(t, client, user.OrganizationID, template.ID)
-		before := database.Now()
 
 		ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
 		defer cancel()
 
-		logs, closer, err := client.WorkspaceBuildLogsAfter(ctx, workspace.LatestBuild.ID, before)
+		logs, closer, err := client.WorkspaceBuildLogsAfter(ctx, workspace.LatestBuild.ID, 0)
 		require.NoError(t, err)
 		defer closer.Close()
 		for {
@@ -121,7 +117,7 @@ func TestProvisionerJobLogs(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
 		defer cancel()
 
-		logs, err := client.WorkspaceBuildLogsBefore(ctx, workspace.LatestBuild.ID, time.Now())
+		logs, err := client.WorkspaceBuildLogsBefore(ctx, workspace.LatestBuild.ID, 0)
 		require.NoError(t, err)
 		require.Greater(t, len(logs), 1)
 	})

coderd/templateversions_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"net/http"
 	"testing"
-	"time"
 
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
@@ -430,7 +429,6 @@ func TestTemplateVersionLogs(t *testing.T) {
 	t.Parallel()
 	client := coderdtest.New(t, &coderdtest.Options{IncludeProvisionerDaemon: true})
 	user := coderdtest.CreateFirstUser(t, client)
-	before := time.Now()
 	version := coderdtest.CreateTemplateVersion(t, client, user.OrganizationID, &echo.Responses{
 		Parse:           echo.ParseComplete,
 		ProvisionDryRun: echo.ProvisionComplete,
@@ -465,7 +463,7 @@ func TestTemplateVersionLogs(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
 	defer cancel()
 
-	logs, closer, err := client.TemplateVersionLogsAfter(ctx, version.ID, before)
+	logs, closer, err := client.TemplateVersionLogsAfter(ctx, version.ID, 0)
 	require.NoError(t, err)
 	defer closer.Close()
 	for {
@@ -625,7 +623,6 @@ func TestTemplateVersionDryRun(t *testing.T) {
 		defer cancel()
 
 		// Create template version dry-run
-		after := time.Now()
 		job, err := client.CreateTemplateVersionDryRun(ctx, version.ID, codersdk.CreateTemplateVersionDryRunRequest{
 			ParameterValues: []codersdk.CreateParameterRequest{},
 		})
@@ -637,7 +634,7 @@ func TestTemplateVersionDryRun(t *testing.T) {
 		require.Equal(t, job.ID, newJob.ID)
 
 		// Stream logs
-		logs, closer, err := client.TemplateVersionDryRunLogsAfter(ctx, version.ID, job.ID, after)
+		logs, closer, err := client.TemplateVersionDryRunLogsAfter(ctx, version.ID, job.ID, 0)
 		require.NoError(t, err)
 		defer closer.Close()
 

coderd/workspacebuilds_test.go

@@ -452,7 +452,6 @@ func TestWorkspaceBuildLogs(t *testing.T) {
 	t.Parallel()
 	client := coderdtest.New(t, &coderdtest.Options{IncludeProvisionerDaemon: true})
 	user := coderdtest.CreateFirstUser(t, client)
-	before := time.Now()
 	version := coderdtest.CreateTemplateVersion(t, client, user.OrganizationID, &echo.Responses{
 		Parse: echo.ParseComplete,
 		Provision: []*proto.Provision_Response{{
@@ -487,7 +486,7 @@ func TestWorkspaceBuildLogs(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitLong)
 	defer cancel()
 
-	logs, closer, err := client.WorkspaceBuildLogsAfter(ctx, workspace.LatestBuild.ID, before.Add(-time.Hour))
+	logs, closer, err := client.WorkspaceBuildLogsAfter(ctx, workspace.LatestBuild.ID, 0)
 	require.NoError(t, err)
 	defer closer.Close()
 	for {