Skip to content

Commit 83c631b

Browse files
cschleidencschleiden
authored
Merge branch 'main' into patch-1
2 parents de0841a + 4b784c4 commit 83c631b

40 files changed

+185
-297
lines changed

‎.github/workflows/bench.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ jobs:
3333
- name: Set up Go
3434
uses: actions/setup-go@v3
3535
with:
36-
go-version: 1.19
36+
go-version: 1.21
3737
check-latest: true
3838
cache: true
3939

‎.github/workflows/go.yml‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ jobs:
2121
- name: Set up Go
2222
uses: actions/setup-go@v3
2323
with:
24-
go-version: 1.19
24+
go-version: 1.21
2525
check-latest: true
2626
cache: true
2727

@@ -51,7 +51,7 @@ jobs:
5151
- name: Set up Go
5252
uses: actions/setup-go@v3
5353
with:
54-
go-version: 1.19
54+
go-version: 1.21
5555
check-latest: true
5656
cache: true
5757

@@ -85,7 +85,7 @@ jobs:
8585
- name: Set up Go
8686
uses: actions/setup-go@v3
8787
with:
88-
go-version: 1.19
88+
go-version: 1.21
8989
check-latest: true
9090
cache: true
9191

@@ -111,7 +111,7 @@ jobs:
111111
- name: Set up Go
112112
uses: actions/setup-go@v3
113113
with:
114-
go-version: 1.19
114+
go-version: 1.21
115115
check-latest: true
116116
cache: true
117117

‎activity/logger.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,12 @@ package activity
22

33
import (
44
"context"
5+
"log/slog"
56

67
"github.com/cschleiden/go-workflows/internal/activity"
7-
"github.com/cschleiden/go-workflows/log"
88
)
99

1010
// Logger returns a logger with the workflow instance this activity is executed for set as default fields
11-
func Logger(ctx context.Context) log.Logger {
11+
func Logger(ctx context.Context) *slog.Logger {
1212
return activity.GetActivityState(ctx).Logger
1313
}

‎activitytester/activitytester.go‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,15 @@ package activitytester
22

33
import (
44
"context"
5+
"log/slog"
56

67
"github.com/cschleiden/go-workflows/internal/activity"
78
"github.com/cschleiden/go-workflows/internal/core"
8-
dlogger "github.com/cschleiden/go-workflows/internal/logger"
9-
"github.com/cschleiden/go-workflows/log"
109
)
1110

12-
func WithActivityTestState(ctx context.Context, activityID, instanceID string, logger log.Logger) context.Context {
11+
func WithActivityTestState(ctx context.Context, activityID, instanceID string, logger *slog.Logger) context.Context {
1312
if logger == nil {
14-
logger = dlogger.NewDefaultLogger()
13+
logger = slog.Default()
1514
}
1615

1716
return activity.WithActivityState(ctx, activity.NewActivityState(activityID, core.NewWorkflowInstance(instanceID, ""), logger))

‎backend/backend.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,13 @@ package backend
33
import (
44
"context"
55
"errors"
6+
"log/slog"
67

78
"github.com/cschleiden/go-workflows/internal/contextpropagation"
89
"github.com/cschleiden/go-workflows/internal/converter"
910
core "github.com/cschleiden/go-workflows/internal/core"
1011
"github.com/cschleiden/go-workflows/internal/history"
1112
"github.com/cschleiden/go-workflows/internal/task"
12-
"github.com/cschleiden/go-workflows/log"
1313
"github.com/cschleiden/go-workflows/metrics"
1414
"github.com/cschleiden/go-workflows/workflow"
1515
"go.opentelemetry.io/otel/trace"
@@ -72,7 +72,7 @@ type Backend interface {
7272
GetStats(ctx context.Context) (*Stats, error)
7373

7474
// Logger returns the configured logger for the backend
75-
Logger() log.Logger
75+
Logger() *slog.Logger
7676

7777
// Tracer returns the configured trace provider for the backend
7878
Tracer() trace.Tracer

‎backend/mock_Backend.go‎

Lines changed: 12 additions & 11 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎backend/mysql/mysql.go‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"encoding/json"
88
"errors"
99
"fmt"
10+
"log/slog"
1011
"strings"
1112
"time"
1213

@@ -17,7 +18,6 @@ import (
1718
"github.com/cschleiden/go-workflows/internal/history"
1819
"github.com/cschleiden/go-workflows/internal/metrickeys"
1920
"github.com/cschleiden/go-workflows/internal/task"
20-
"github.com/cschleiden/go-workflows/log"
2121
"github.com/cschleiden/go-workflows/metrics"
2222
"github.com/cschleiden/go-workflows/workflow"
2323
_ "github.com/go-sql-driver/mysql"
@@ -63,7 +63,7 @@ type mysqlBackend struct {
6363
options backend.Options
6464
}
6565

66-
func (b *mysqlBackend) Logger() log.Logger {
66+
func (b *mysqlBackend) Logger() *slog.Logger {
6767
return b.options.Logger
6868
}
6969

@@ -332,16 +332,17 @@ func (b *mysqlBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, err
332332
FROM instances i
333333
INNER JOIN pending_events pe ON i.instance_id = pe.instance_id
334334
WHERE
335-
i.completed_at IS NULL
335+
state = ? AND i.completed_at IS NULL
336336
AND (pe.visible_at IS NULL OR pe.visible_at <= ?)
337337
AND (i.locked_until IS NULL OR i.locked_until < ?)
338338
AND (i.sticky_until IS NULL OR i.sticky_until < ? OR i.worker = ?)
339339
LIMIT 1
340340
FOR UPDATE OF i SKIP LOCKED`,
341-
now, // event.visible_at
342-
now, // locked_until
343-
now, // sticky_until
344-
b.workerName, // worker
341+
core.WorkflowInstanceStateActive, // state
342+
now, // event.visible_at
343+
now, // locked_until
344+
now, // sticky_until
345+
b.workerName, // worker
345346
)
346347

347348
var id int
@@ -484,7 +485,7 @@ func (b *mysqlBackend) CompleteWorkflowTask(
484485

485486
// Unlock instance, but keep it sticky to the current worker
486487
var completedAt *time.Time
487-
if state == core.WorkflowInstanceStateFinished {
488+
if state == core.WorkflowInstanceStateContinuedAsNew || state == core.WorkflowInstanceStateFinished {
488489
t := time.Now()
489490
completedAt = &t
490491
}

‎backend/options.go‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,21 +1,20 @@
11
package backend
22

33
import (
4+
"log/slog"
45
"time"
56

67
"github.com/cschleiden/go-workflows/internal/contextpropagation"
78
"github.com/cschleiden/go-workflows/internal/converter"
8-
"github.com/cschleiden/go-workflows/internal/logger"
99
mi "github.com/cschleiden/go-workflows/internal/metrics"
1010
"github.com/cschleiden/go-workflows/internal/tracing"
11-
"github.com/cschleiden/go-workflows/log"
1211
"github.com/cschleiden/go-workflows/metrics"
1312
"github.com/cschleiden/go-workflows/workflow"
1413
"go.opentelemetry.io/otel/trace"
1514
)
1615

1716
type Options struct {
18-
Logger log.Logger
17+
Logger *slog.Logger
1918

2019
Metrics metrics.Client
2120

@@ -46,7 +45,7 @@ var DefaultOptions Options = Options{
4645
WorkflowLockTimeout: time.Minute,
4746
ActivityLockTimeout: time.Minute * 2,
4847

49-
Logger: logger.NewDefaultLogger(),
48+
Logger: slog.Default(),
5049
Metrics: mi.NewNoopMetricsClient(),
5150
TracerProvider: trace.NewNoopTracerProvider(),
5251
Converter: converter.DefaultConverter,
@@ -62,7 +61,7 @@ func WithStickyTimeout(timeout time.Duration) BackendOption {
6261
}
6362
}
6463

65-
func WithLogger(logger log.Logger) BackendOption {
64+
func WithLogger(logger *slog.Logger) BackendOption {
6665
return func(o *Options) {
6766
o.Logger = logger
6867
}
@@ -100,7 +99,7 @@ func ApplyOptions(opts ...BackendOption) Options {
10099
}
101100

102101
if options.Logger == nil {
103-
options.Logger = logger.NewDefaultLogger()
102+
options.Logger = slog.Default()
104103
}
105104

106105
return options

‎backend/redis/redis.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package redis
33
import (
44
"context"
55
"fmt"
6+
"log/slog"
67
"time"
78

89
"github.com/cschleiden/go-workflows/backend"
@@ -11,7 +12,6 @@ import (
1112
"github.com/cschleiden/go-workflows/internal/core"
1213
"github.com/cschleiden/go-workflows/internal/history"
1314
"github.com/cschleiden/go-workflows/internal/metrickeys"
14-
"github.com/cschleiden/go-workflows/log"
1515
"github.com/cschleiden/go-workflows/metrics"
1616
"github.com/redis/go-redis/v9"
1717
"go.opentelemetry.io/otel/trace"
@@ -86,7 +86,7 @@ type activityData struct {
8686
Event *history.Event `json:"event,omitempty"`
8787
}
8888

89-
func (rb *redisBackend) Logger() log.Logger {
89+
func (rb *redisBackend) Logger() *slog.Logger {
9090
return rb.options.Logger
9191
}
9292

‎backend/sqlite/sqlite.go‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"encoding/json"
88
"errors"
99
"fmt"
10+
"log/slog"
1011
"strings"
1112
"time"
1213

@@ -17,7 +18,6 @@ import (
1718
"github.com/cschleiden/go-workflows/internal/history"
1819
"github.com/cschleiden/go-workflows/internal/metrickeys"
1920
"github.com/cschleiden/go-workflows/internal/task"
20-
"github.com/cschleiden/go-workflows/log"
2121
"github.com/cschleiden/go-workflows/metrics"
2222
"github.com/cschleiden/go-workflows/workflow"
2323
"github.com/google/uuid"
@@ -67,7 +67,7 @@ type sqliteBackend struct {
6767

6868
var _ backend.Backend = (*sqliteBackend)(nil)
6969

70-
func (sb *sqliteBackend) Logger() log.Logger {
70+
func (sb *sqliteBackend) Logger() *slog.Logger {
7171
return sb.options.Logger
7272
}
7373

@@ -288,7 +288,7 @@ func (sb *sqliteBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, e
288288
WHERE
289289
(locked_until IS NULL OR locked_until < ?)
290290
AND (sticky_until IS NULL OR sticky_until < ? OR worker = ?)
291-
AND completed_at IS NULL
291+
AND state = ? AND i.completed_at IS NULL
292292
AND EXISTS (
293293
SELECT 1
294294
FROM pending_events
@@ -298,10 +298,11 @@ func (sb *sqliteBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, e
298298
) RETURNING id, execution_id, parent_instance_id, parent_execution_id, parent_schedule_event_id, metadata, sticky_until`,
299299
now.Add(sb.options.WorkflowLockTimeout), // new locked_until
300300
sb.workerName,
301-
now, // locked_until
302-
now, // sticky_until
303-
sb.workerName, // worker
304-
now, // event.visible_at
301+
now, // locked_until
302+
now, // sticky_until
303+
sb.workerName, // worker
304+
core.WorkflowInstanceStateActive, // state
305+
now, // pending_event.visible_at
305306
)
306307

307308
var instanceID, executionID string
@@ -383,7 +384,7 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
383384
defer tx.Rollback()
384385

385386
var completedAt *time.Time
386-
if state == core.WorkflowInstanceStateFinished || state == core.WorkflowInstanceStateContinuedAsNew {
387+
if state == core.WorkflowInstanceStateContinuedAsNew || state == core.WorkflowInstanceStateFinished {
387388
t := time.Now()
388389
completedAt = &t
389390
}

0 commit comments

Comments
 (0)