Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Refactor coroutine implementation #87

Merged
merged 3 commits into from
Oct 19, 2023
Merged

Refactor coroutine implementation #87

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7f2d86e
Refactor coroutine implementation
dfarr Oct 18, 2023
a419b8c
Remove echo endpoint
dfarr Oct 18, 2023
4ee4cb8
Move coroutines between two slices
dfarr Oct 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
254 changes: 125 additions & 129 deletions internal/app/coroutines/cancelPromise.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,16 +10,16 @@
"github.com/resonatehq/resonate/pkg/promise"
)

func CancelPromise(req *t_api.Request, res func(*t_api.Response, error)) *scheduler.Coroutine {
return scheduler.NewCoroutine("CancelPromise", func(s *scheduler.Scheduler, c *scheduler.Coroutine) {
func CancelPromise(req *t_api.Request, res func(*t_api.Response, error)) *scheduler.Coroutine[*t_aio.Completion, *t_aio.Submission] {
return scheduler.NewCoroutine("CancelPromise", func(c *scheduler.Coroutine[*t_aio.Completion, *t_aio.Submission]) {
if req.CancelPromise.Value.Headers == nil {
req.CancelPromise.Value.Headers = map[string]string{}
}
if req.CancelPromise.Value.Data == nil {
req.CancelPromise.Value.Data = []byte{}
}

submission := &t_aio.Submission{
completion, err := c.Yield(&t_aio.Submission{
Kind: t_aio.Store,
Store: &t_aio.StoreSubmission{
Transaction: &t_aio.Transaction{
Expand All @@ -33,154 +33,150 @@
},
},
},
})

if err != nil {
slog.Error("failed to read promise", "req", req, "err", err)
res(nil, err)
return

Check warning on line 41 in internal/app/coroutines/cancelPromise.go

View check run for this annotation

Codecov / codecov/patch

internal/app/coroutines/cancelPromise.go#L39-L41 

Added lines #L39 - L41 were not covered by tests
}

c.Yield(submission, func(completion *t_aio.Completion, err error) {
util.Assert(completion.Store != nil, "completion must not be nil")

result := completion.Store.Results[0].ReadPromise
util.Assert(result.RowsReturned == 0 || result.RowsReturned == 1, "result must return 0 or 1 rows")

if result.RowsReturned == 0 {
res(&t_api.Response{
Kind: t_api.CancelPromise,
CancelPromise: &t_api.CancelPromiseResponse{
Status: t_api.ResponseNotFound,
},
}, nil)
} else {
p, err := result.Records[0].Promise()
if err != nil {
slog.Error("failed to read promise", "req", req, "err", err)
slog.Error("failed to parse promise record", "record", result.Records[0], "err", err)

Check warning on line 59 in internal/app/coroutines/cancelPromise.go

View check run for this annotation

Codecov / codecov/patch

internal/app/coroutines/cancelPromise.go#L59 

Added line #L59 was not covered by tests
res(nil, err)
return
}

util.Assert(completion.Store != nil, "completion must not be nil")

result := completion.Store.Results[0].ReadPromise
util.Assert(result.RowsReturned == 0 || result.RowsReturned == 1, "result must return 0 or 1 rows")

if result.RowsReturned == 0 {
res(&t_api.Response{
Kind: t_api.CancelPromise,
CancelPromise: &t_api.CancelPromiseResponse{
Status: t_api.ResponseNotFound,
},
}, nil)
} else {
p, err := result.Records[0].Promise()
if err != nil {
slog.Error("failed to parse promise record", "record", result.Records[0], "err", err)
res(nil, err)
return
}
if p.State == promise.Pending {
if c.Time() >= p.Timeout {
c.Scheduler.Add(TimeoutPromise(p, CancelPromise(req, res), func(err error) {
if err != nil {
slog.Error("failed to timeout promise", "req", req, "err", err)
res(nil, err)
return
}

Check warning on line 71 in internal/app/coroutines/cancelPromise.go

View check run for this annotation

Codecov / codecov/patch

internal/app/coroutines/cancelPromise.go#L67-L71 

Added lines #L67 - L71 were not covered by tests

if p.State == promise.Pending {
if s.Time() >= p.Timeout {
s.Add(TimeoutPromise(p, CancelPromise(req, res), func(err error) {
if err != nil {
slog.Error("failed to timeout promise", "req", req, "err", err)
res(nil, err)
return
}

res(&t_api.Response{
Kind: t_api.CancelPromise,
CancelPromise: &t_api.CancelPromiseResponse{
Status: t_api.ResponseForbidden,
Promise: &promise.Promise{
Id: p.Id,
State: promise.Timedout,
Param: p.Param,
Value: promise.Value{
Headers: map[string]string{},
Data: []byte{},
},
Timeout: p.Timeout,
IdempotencyKeyForCreate: p.IdempotencyKeyForCreate,
IdempotencyKeyForComplete: p.IdempotencyKeyForComplete,
Tags: p.Tags,
CreatedOn: p.CreatedOn,
CompletedOn: &p.Timeout,
res(&t_api.Response{
Kind: t_api.CancelPromise,
CancelPromise: &t_api.CancelPromiseResponse{
Status: t_api.ResponseForbidden,
Promise: &promise.Promise{
Id: p.Id,
State: promise.Timedout,
Param: p.Param,
Value: promise.Value{
Headers: map[string]string{},
Data: []byte{},

Check warning on line 83 in internal/app/coroutines/cancelPromise.go

View check run for this annotation

Codecov / codecov/patch

internal/app/coroutines/cancelPromise.go#L73-L83 

Added lines #L73 - L83 were not covered by tests
},
Timeout: p.Timeout,
IdempotencyKeyForCreate: p.IdempotencyKeyForCreate,
IdempotencyKeyForComplete: p.IdempotencyKeyForComplete,
Tags: p.Tags,
CreatedOn: p.CreatedOn,
CompletedOn: &p.Timeout,

Check warning on line 90 in internal/app/coroutines/cancelPromise.go

View check run for this annotation

Codecov / codecov/patch

internal/app/coroutines/cancelPromise.go#L85-L90 

Added lines #L85 - L90 were not covered by tests
},
}, nil)
}))
} else {
completedOn := s.Time()
submission := &t_aio.Submission{
Kind: t_aio.Store,
Store: &t_aio.StoreSubmission{
Transaction: &t_aio.Transaction{
Commands: []*t_aio.Command{
{
Kind: t_aio.UpdatePromise,
UpdatePromise: &t_aio.UpdatePromiseCommand{
Id: req.CancelPromise.Id,
State: promise.Canceled,
Value: req.CancelPromise.Value,
IdempotencyKey: req.CancelPromise.IdempotencyKey,
CompletedOn: completedOn,
},
},
}, nil)

Check warning on line 93 in internal/app/coroutines/cancelPromise.go

View check run for this annotation

Codecov / codecov/patch

internal/app/coroutines/cancelPromise.go#L92-L93 

Added lines #L92 - L93 were not covered by tests
}))
} else {
completedOn := c.Time()
completion, err := c.Yield(&t_aio.Submission{
Kind: t_aio.Store,
Store: &t_aio.StoreSubmission{
Transaction: &t_aio.Transaction{
Commands: []*t_aio.Command{
{
Kind: t_aio.UpdatePromise,
UpdatePromise: &t_aio.UpdatePromiseCommand{
Id: req.CancelPromise.Id,
State: promise.Canceled,
Value: req.CancelPromise.Value,
IdempotencyKey: req.CancelPromise.IdempotencyKey,
CompletedOn: completedOn,
},
{
Kind: t_aio.CreateNotifications,
CreateNotifications: &t_aio.CreateNotificationsCommand{
PromiseId: req.CancelPromise.Id,
Time: completedOn,
},
},
{
Kind: t_aio.CreateNotifications,
CreateNotifications: &t_aio.CreateNotificationsCommand{
PromiseId: req.CancelPromise.Id,
Time: completedOn,
},
{
Kind: t_aio.DeleteSubscriptions,
DeleteSubscriptions: &t_aio.DeleteSubscriptionsCommand{
PromiseId: req.CancelPromise.Id,
},
},
{
Kind: t_aio.DeleteSubscriptions,
DeleteSubscriptions: &t_aio.DeleteSubscriptionsCommand{
PromiseId: req.CancelPromise.Id,
},
},
},
},
}
},
})

c.Yield(submission, func(completion *t_aio.Completion, err error) {
if err != nil {
slog.Error("failed to update promise", "req", req, "err", err)
res(nil, err)
return
}

util.Assert(completion.Store != nil, "completion must not be nil")

result := completion.Store.Results[0].UpdatePromise
util.Assert(result.RowsAffected == 0 || result.RowsAffected == 1, "result must return 0 or 1 rows")

if result.RowsAffected == 1 {
res(&t_api.Response{
Kind: t_api.CancelPromise,
CancelPromise: &t_api.CancelPromiseResponse{
Status: t_api.ResponseCreated,
Promise: &promise.Promise{
Id: p.Id,
State: promise.Canceled,
Param: p.Param,
Value: req.CancelPromise.Value,
Timeout: p.Timeout,
IdempotencyKeyForCreate: p.IdempotencyKeyForCreate,
IdempotencyKeyForComplete: req.CancelPromise.IdempotencyKey,
Tags: p.Tags,
CreatedOn: p.CreatedOn,
CompletedOn: &completedOn,
},
},
}, nil)
} else {
s.Add(CancelPromise(req, res))
}
})
if err != nil {
slog.Error("failed to update promise", "req", req, "err", err)
res(nil, err)
return

Check warning on line 133 in internal/app/coroutines/cancelPromise.go

View check run for this annotation

Codecov / codecov/patch

internal/app/coroutines/cancelPromise.go#L131-L133 

Added lines #L131 - L133 were not covered by tests
}
} else {
status := t_api.ResponseForbidden
strict := req.CancelPromise.Strict && p.State != promise.Canceled

if !strict && p.IdempotencyKeyForComplete.Match(req.CancelPromise.IdempotencyKey) {
status = t_api.ResponseOK
util.Assert(completion.Store != nil, "completion must not be nil")

result := completion.Store.Results[0].UpdatePromise
util.Assert(result.RowsAffected == 0 || result.RowsAffected == 1, "result must return 0 or 1 rows")

if result.RowsAffected == 1 {
res(&t_api.Response{
Kind: t_api.CancelPromise,
CancelPromise: &t_api.CancelPromiseResponse{
Status: t_api.ResponseCreated,
Promise: &promise.Promise{
Id: p.Id,
State: promise.Canceled,
Param: p.Param,
Value: req.CancelPromise.Value,
Timeout: p.Timeout,
IdempotencyKeyForCreate: p.IdempotencyKeyForCreate,
IdempotencyKeyForComplete: req.CancelPromise.IdempotencyKey,
Tags: p.Tags,
CreatedOn: p.CreatedOn,
CompletedOn: &completedOn,
},
},
}, nil)
} else {
c.Scheduler.Add(CancelPromise(req, res))
}
}
} else {
status := t_api.ResponseForbidden
strict := req.CancelPromise.Strict && p.State != promise.Canceled

res(&t_api.Response{
Kind: t_api.CancelPromise,
CancelPromise: &t_api.CancelPromiseResponse{
Status: status,
Promise: p,
},
}, nil)
if !strict && p.IdempotencyKeyForComplete.Match(req.CancelPromise.IdempotencyKey) {
status = t_api.ResponseOK
}

res(&t_api.Response{
Kind: t_api.CancelPromise,
CancelPromise: &t_api.CancelPromiseResponse{
Status: status,
Promise: p,
},
}, nil)
}
})
}
})
}
Loading