poc

makasim · Nov 11, 2024 · a82a7c5 · a82a7c5
1 parent 8962421
commit a82a7c5
Show file tree

Hide file tree

Showing 45 changed files with 112 additions and 69 deletions.
diff --git a/engine.go b/engine.go
@@ -86,7 +86,7 @@ func (e *Engine) Execute(stateCtx *StateCtx) error {
 		conflictErr := &ErrCommitConflict{}
 
 		if err = e.do(cmd0); errors.As(err, conflictErr) {
-			e.l.Info("do conflict",
+			e.l.Info("engine: do conflict",
 				"sess", cmd0.SessID(),
 				"conflict", err.Error(),
 				"id", stateCtx.Current.ID,

diff --git a/log.go b/log.go
@@ -27,17 +27,14 @@ func logExecute(stateCtx *StateCtx, l *slog.Logger) {
 		args = append(args, "recovered", currTs.Annotations[RecoveryAttemptAnnotation])
 	}
 
-	l.Info("execute flow", args...)
+	l.Info("engine: execute", args...)
 }
 
 func logDo(cmd0 Command, l *slog.Logger) {
 	args := []any{"sess", cmd0.SessID()}
 
 	switch cmd := cmd0.(type) {
 	case *CommitCommand:
-		//for _, c := range cmd.Commands {
-		//	log.Printf("commit: %T", c)
-		//}
 		args = append(args, "cmd", "commit", "len", len(cmd.Commands))
 	case *CommitStateCtxCommand:
 		args = append(args, "cmd", "commit_state_ctx", "id", cmd.StateCtx.Current.ID, "rev", cmd.StateCtx.Current.Rev)
@@ -92,6 +89,6 @@ func logDo(cmd0 Command, l *slog.Logger) {
 		args = append(args, "cmd", fmt.Sprintf("%T", cmd))
 	}
 
-	l.Info("do commands", args...)
+	l.Info("engine: do", args...)
 
 }
diff --git a/memdriver/delayer.go b/memdriver/delayer.go
@@ -4,7 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log"
+	"log/slog"
 	"sync"
 	"time"
 
@@ -17,11 +17,13 @@ type Delayer struct {
 	e      *flowstate.Engine
 	stopCh chan struct{}
 	wg     sync.WaitGroup
+	l      *slog.Logger
 }
 
-func NewDelayer() *Delayer {
+func NewDelayer(l *slog.Logger) *Delayer {
 	return &Delayer{
 		stopCh: make(chan struct{}),
+		l:      l,
 	}
 }
 
@@ -43,26 +45,46 @@ func (d *Delayer) Do(cmd0 flowstate.Command) error {
 		t := time.NewTimer(cmd.Duration)
 		defer t.Stop()
 
+		stateCtx := cmd.DelayStateCtx
+
 		select {
 		case <-t.C:
-			if cmd.DelayStateCtx.Current.Transition.Annotations[flowstate.DelayCommitAnnotation] == `true` {
+			if stateCtx.Current.Transition.Annotations[flowstate.DelayCommitAnnotation] == `true` {
 				conflictErr := &flowstate.ErrCommitConflict{}
 				if err := d.e.Do(flowstate.Commit(
 					flowstate.CommitStateCtx(cmd.DelayStateCtx),
 				)); errors.As(err, conflictErr) {
-					log.Printf(`ERROR: memdriver: delayer: engine: commit: %s\n`, conflictErr)
+					d.l.Info("delayer: commit conflict",
+						"sess", cmd.SessID(),
+						"conflict", err.Error(),
+						"id", stateCtx.Current.ID,
+						"rev", stateCtx.Current.Rev,
+					)
 					return
 				} else if err != nil {
-					log.Printf(`ERROR: memdriver: delayer: engine: commit: %s`, err)
+					d.l.Error("delayer: commit failed",
+						"sess", cmd.SessID(),
+						"error", err,
+						"id", stateCtx.Current.ID,
+						"rev", stateCtx.Current.Rev,
+					)
 					return
 				}
 			}
 
-			if err := d.e.Execute(cmd.DelayStateCtx); err != nil {
-				log.Printf(`ERROR: memdriver: delayer: engine: execute: %s`, err)
+			if err := d.e.Execute(stateCtx); err != nil {
+				d.l.Error("delayer: execute failed",
+					"sess", stateCtx.SessID(),
+					"error", err,
+					"id", stateCtx.Current.ID,
+					"rev", stateCtx.Current.Rev,
+				)
 			}
 		case <-d.stopCh:
-			log.Printf(`ERROR: memdriver: delayer: state delay was terminated`)
+			d.l.Error("delayer: delaying terminated",
+				"id", stateCtx.Current.ID,
+				"rev", stateCtx.Current.Rev,
+			)
 			return
 		}
 	}()

diff --git a/memdriver/driver.go b/memdriver/driver.go
@@ -4,24 +4,34 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
+	"os"
 
 	"github.com/makasim/flowstate"
 )
 
 type Driver struct {
 	*FlowRegistry
 
-	l         *Log
+	log       *Log
 	doers     []flowstate.Doer
 	recoverer flowstate.Doer
+
+	l *slog.Logger
 }
 
 func New(opts ...Option) *Driver {
-	l := &Log{}
+	log := &Log{}
 
 	d := &Driver{
-		l:            l,
+		log:          log,
 		FlowRegistry: &FlowRegistry{},
+
+		l: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{})),
+	}
+
+	for _, opt := range opts {
+		opt(d)
 	}
 
 	doers := []flowstate.Doer{
@@ -37,17 +47,13 @@ func New(opts ...Option) *Driver {
 
 		NewDataLog(),
 		NewFlowGetter(d.FlowRegistry),
-		NewCommiter(l),
-		NewGetter(l),
-		NewWatcher(l),
-		NewDelayer(),
+		NewCommiter(log),
+		NewGetter(log),
+		NewWatcher(log),
+		NewDelayer(d.l),
 	}
 	d.doers = doers
 
-	for _, opt := range opts {
-		opt(d)
-	}
-
 	return d
 }
 
@@ -105,3 +111,9 @@ func WithRecoverer(r flowstate.Doer) Option {
 		d.recoverer = r
 	}
 }
+
+func WithLogger(l *slog.Logger) Option {
+	return func(d *Driver) {
+		d.l = l
+	}
+}
diff --git a/memdriver/suite_test.go b/memdriver/suite_test.go
@@ -10,7 +10,8 @@ import (
 
 func TestSuite(t *testing.T) {
 	s := testcases.Get(func(t testcases.TestingT) (flowstate.Doer, testcases.FlowRegistry) {
-		d := memdriver.New()
+		l, _ := testcases.NewTestLogger(t)
+		d := memdriver.New(memdriver.WithLogger(l))
 		return d, d
 	})
 

diff --git a/pgdriver/driver.go b/pgdriver/driver.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
+	"os"
 	"time"
 
 	"github.com/makasim/flowstate"
@@ -18,6 +20,7 @@ type Driver struct {
 	doers []flowstate.Doer
 
 	recoverer flowstate.Doer
+	l         *slog.Logger
 }
 
 func New(conn conn, opts ...Option) *Driver {
@@ -26,6 +29,11 @@ func New(conn conn, opts ...Option) *Driver {
 
 		q:            &queries{},
 		FlowRegistry: &memdriver.FlowRegistry{},
+		l:            slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{})),
+	}
+
+	for _, opt := range opts {
+		opt(d)
 	}
 
 	d.doers = []flowstate.Doer{
@@ -50,10 +58,6 @@ func New(conn conn, opts ...Option) *Driver {
 		NewDelayer(d.conn, d.q, time.Now),
 	}
 
-	for _, opt := range opts {
-		opt(d)
-	}
-
 	return d
 }
 
@@ -111,3 +115,9 @@ func WithRecoverer(r flowstate.Doer) Option {
 		d.recoverer = r
 	}
 }
+
+func WithLogger(l *slog.Logger) Option {
+	return func(d *Driver) {
+		d.l = l
+	}
+}
diff --git a/pgdriver/suite_test.go b/pgdriver/suite_test.go
@@ -32,7 +32,8 @@ func TestSuite(t *testing.T) {
 			conn.Close()
 		})
 
-		d := pgdriver.New(conn)
+		l, _ := testcases.NewTestLogger(t)
+		d := pgdriver.New(conn, pgdriver.WithLogger(l))
 		return d, d
 	})
 

diff --git a/testcases/actor.go b/testcases/actor.go
@@ -49,7 +49,7 @@ func Actor(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return nil, fmt.Errorf("must never be executed")
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/call_flow.go b/testcases/call_flow.go
@@ -79,7 +79,7 @@ func CallFlow(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return flowstate.End(stateCtx), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/call_flow_with_commit.go b/testcases/call_flow_with_commit.go
@@ -82,7 +82,7 @@ func CallFlowWithCommit(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return flowstate.End(stateCtx), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/call_flow_with_watch.go b/testcases/call_flow_with_watch.go
@@ -97,7 +97,7 @@ func CallFlowWithWatch(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/condition.go b/testcases/condition.go
@@ -33,7 +33,7 @@ func Condition(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return flowstate.End(stateCtx), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/cron.go b/testcases/cron.go
@@ -87,7 +87,7 @@ func Cron(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/data_flow_config.go b/testcases/data_flow_config.go
@@ -125,7 +125,7 @@ func DataFlowConfig(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return flowstate.End(stateCtx), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/data_store_get.go b/testcases/data_store_get.go
@@ -63,7 +63,7 @@ func DataStoreGet(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return flowstate.End(stateCtx), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/data_store_get_with_commit.go b/testcases/data_store_get_with_commit.go
@@ -63,7 +63,7 @@ func DataStoreGetWithCommit(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return flowstate.End(stateCtx), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/delay_delayed_win_with_commit.go b/testcases/delay_delayed_win_with_commit.go
@@ -48,7 +48,7 @@ func Delay_DelayedWin_WithCommit(t TestingT, d flowstate.Doer, fr FlowRegistry)
 		), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/delay_engine_do.go b/testcases/delay_engine_do.go
@@ -33,7 +33,7 @@ func Delay_EngineDo(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return flowstate.End(stateCtx), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/delay_paused.go b/testcases/delay_paused.go
@@ -38,7 +38,7 @@ func Delay_Paused(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/delay_paused_with_commit.go b/testcases/delay_paused_with_commit.go
@@ -40,7 +40,7 @@ func Delay_PausedWithCommit(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/delay_return.go b/testcases/delay_return.go
@@ -27,7 +27,7 @@ func Delay_Return(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return flowstate.End(stateCtx), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/delay_transited_win_with_commit.go b/testcases/delay_transited_win_with_commit.go
@@ -44,7 +44,7 @@ func Delay_TransitedWin_WithCommit(t TestingT, d flowstate.Doer, fr FlowRegistry
 		), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {

diff --git a/testcases/fork.go b/testcases/fork.go
@@ -51,7 +51,7 @@ func Fork(t TestingT, d flowstate.Doer, fr FlowRegistry) {
 		return flowstate.End(stateCtx), nil
 	}))
 
-	l, _ := newTestLogger(t)
+	l, _ := NewTestLogger(t)
 	e, err := flowstate.NewEngine(d, l)
 	require.NoError(t, err)
 	defer func() {