210 lines
8.8 KiB
Go
210 lines
8.8 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"time"
|
|
|
|
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
"git.yuntuo.hk/k2safe/orbit/weave/definition"
|
|
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
|
weave "git.yuntuo.hk/k2safe/orbit/weave/model"
|
|
"git.yuntuo.hk/k2safe/orbit/weave/signal"
|
|
)
|
|
|
|
type WorkflowSignalStore struct{ *TaskStore }
|
|
|
|
func NewWorkflowSignalStore(db *sql.DB) *WorkflowSignalStore {
|
|
return &WorkflowSignalStore{NewTaskStore(db)}
|
|
}
|
|
|
|
var _ signal.Store = (*WorkflowSignalStore)(nil)
|
|
|
|
func signalContract(node definition.Step) *definition.Signal {
|
|
if node.Signal != nil {
|
|
return node.Signal
|
|
}
|
|
if node.Poll != nil {
|
|
return node.Poll.FinalSignal
|
|
}
|
|
return nil
|
|
}
|
|
|
|
const signalColumns = `id,namespace_id,workflow_id,revision,type,COALESCE(payload_version,''),target_step_key,COALESCE(target_activation_sequence,0),correlation_key,payload,actor_id,COALESCE(evidence_ref,''),state,COALESCE(disposition,''),accepted_at,expires_at,consumed_wait_id`
|
|
|
|
func readWorkflowSignal(ctx context.Context, tx *sql.Tx, ns, workflow, id core.ID) (signal.View, error) {
|
|
var v signal.View
|
|
var payload []byte
|
|
err := tx.QueryRowContext(ctx, `SELECT `+signalColumns+` FROM weave_signals WHERE namespace_id=? AND workflow_id=? AND id=?`, ns, workflow, id).Scan(&v.ID, &v.NamespaceID, &v.WorkflowID, &v.Revision, &v.Name, &v.PayloadVersion, &v.TargetStepKey, &v.TargetActivationSequence, &v.CorrelationKey, &payload, &v.ActorID, &v.EvidenceRef, &v.State, &v.Disposition, &v.AcceptedAt, &v.ExpiresAt, &v.ConsumedWaitID)
|
|
v.Payload = payload
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
err = fault.NotFound
|
|
}
|
|
return v, err
|
|
}
|
|
|
|
func (s *WorkflowSignalStore) Send(ctx context.Context, t auth.Token, workflow core.ID, key string, in signal.Input) (signal.Result, error) {
|
|
var out signal.Result
|
|
err := s.transaction(ctx, t, signal.SendAction, func(tx *sql.Tx, p auth.Permissions) error {
|
|
if _, err := lockApplicationResources(ctx, tx, t.NamespaceID, false); err != nil {
|
|
return err
|
|
}
|
|
id, err := core.NewID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "workflow.signal", string(workflow), key, "workflow_signal", id, in)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
g, _, err := lockWorkflowGraph(ctx, tx, t.NamespaceID, workflow, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, queue := range g.version.Analysis.Queues {
|
|
if !p.AllowsQueue(queue) {
|
|
return fault.NotFound
|
|
}
|
|
}
|
|
node, ok := g.program.Step(in.TargetStepKey)
|
|
contract := signalContract(node)
|
|
if !ok || contract == nil || contract.Name != in.Name || contract.PayloadVersion != in.PayloadVersion || definition.ValidateData(contract.PayloadType, in.Payload) != nil {
|
|
return fault.Invalid("signal does not match fixed workflow contract")
|
|
}
|
|
if node.Poll != nil {
|
|
outcome, err := g.program.ClassifyPoll(in.TargetStepKey, in.Payload)
|
|
if err != nil || outcome == engine.PollPending {
|
|
return fault.Invalid("final signal requires a terminal business result")
|
|
}
|
|
}
|
|
if cached != nil {
|
|
if err := json.Unmarshal(cached, &out); err != nil {
|
|
return err
|
|
}
|
|
current, err := readWorkflowSignal(ctx, tx, t.NamespaceID, workflow, out.Signal.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if out.Signal.NamespaceID != t.NamespaceID || out.Signal.WorkflowID != workflow || !current.AcceptedAt.Equal(out.Signal.AcceptedAt) {
|
|
return errors.New("workflow signal receipt mismatch")
|
|
}
|
|
out.Deduplicated = true
|
|
return nil
|
|
}
|
|
var count int
|
|
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM (SELECT id FROM weave_signals WHERE namespace_id=? AND workflow_id=? LIMIT ?) admitted_signals`, t.NamespaceID, workflow, signal.MaxPerWorkflow).Scan(&count); err != nil {
|
|
return err
|
|
}
|
|
if count >= signal.MaxPerWorkflow {
|
|
return fault.New("WORKFLOW_SIGNAL_LIMIT", "workflow signal limit reached")
|
|
}
|
|
row := g.steps[in.TargetStepKey]
|
|
sequence := uint32(1)
|
|
if row.view.ActivationID != nil {
|
|
if err := tx.QueryRowContext(ctx, `SELECT sequence FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? AND id=?`, t.NamespaceID, workflow, row.view.ID, row.view.ActivationID).Scan(&sequence); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if in.TargetActivationSequence > sequence {
|
|
return fault.Conflict
|
|
}
|
|
now, err := dbNow(ctx, tx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
state, disposition := weave.SignalState("PENDING"), ""
|
|
switch {
|
|
case in.TargetActivationSequence < sequence:
|
|
state, disposition = "REJECTED", "STALE_ACTIVATION"
|
|
case g.view.State.Terminal():
|
|
state, disposition = "REJECTED", "WORKFLOW_SETTLED"
|
|
case g.view.StopKind != nil:
|
|
state, disposition = "REJECTED", "WORKFLOW_STOPPING"
|
|
case !now.Before(g.view.DeadlineAt):
|
|
state, disposition = "EXPIRED", "WORKFLOW_DEADLINE"
|
|
case row.view.State == weave.StepSkipped || row.view.State == weave.StepSucceeded || row.view.State == weave.StepFailed || row.view.State == weave.StepCancelled || row.view.State == weave.StepNeedsReview:
|
|
state, disposition = "REJECTED", "STEP_SETTLED"
|
|
case row.view.State == weave.StepBlocked && g.view.CancelRequestedAt != nil:
|
|
state, disposition = "REJECTED", "WORKFLOW_CANCELLING"
|
|
}
|
|
if disposition == "" && row.view.ActivationID != nil {
|
|
var name, version *string
|
|
var correlation, waitState string
|
|
var deadline time.Time
|
|
if err := tx.QueryRowContext(ctx, `SELECT signal_name,payload_version,correlation_key,state,deadline_at FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=? FOR UPDATE`, t.NamespaceID, workflow, row.view.ActivationID).Scan(&name, &version, &correlation, &waitState, &deadline); err != nil {
|
|
return err
|
|
}
|
|
if name == nil || version == nil || *name != in.Name || *version != in.PayloadVersion {
|
|
return errors.New("workflow signal wait contract invalid")
|
|
}
|
|
switch {
|
|
case waitState != "WAITING":
|
|
state, disposition = "REJECTED", "WAIT_SETTLED"
|
|
case correlation != in.CorrelationKey:
|
|
state, disposition = "REJECTED", "CORRELATION_MISMATCH"
|
|
case !now.Before(deadline):
|
|
state, disposition = "EXPIRED", "SIGNAL_DEADLINE"
|
|
}
|
|
}
|
|
digest, err := requestDigest(in)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
keyDigest := core.Digest(sha256.Sum256([]byte(key)))
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO weave_signals(namespace_id,id,workflow_id,key_digest,request_digest,type,payload_version,target_step_key,target_activation_sequence,correlation_key,payload,actor_id,evidence_ref,state,disposition,accepted_at,expires_at,available_at,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, t.NamespaceID, id, workflow, keyDigest, digest, in.Name, in.PayloadVersion, in.TargetStepKey, in.TargetActivationSequence, []byte(in.CorrelationKey), string(in.Payload), string(t.CredentialID), businessKeyValue(in.EvidenceRef), state, businessKeyValue(disposition), now, g.view.DeadlineAt, g.view.DeadlineAt, now, now)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if state == "PENDING" && row.view.ActivationID != nil {
|
|
if err := saveWorkflowClock(ctx, tx, g, row, string(row.view.Kind), row.view.Revision, now); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
res, err := tx.ExecContext(ctx, `UPDATE weave_instances SET revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, now, t.NamespaceID, workflow, g.view.Revision)
|
|
if err := exactlyOne(res, err); err != nil {
|
|
return err
|
|
}
|
|
g.view.Revision++
|
|
meta := map[string]any{"signal_id": id, "workflow_id": workflow, "target_step_key": in.TargetStepKey, "target_activation_sequence": in.TargetActivationSequence, "state": state, "disposition": disposition, "request_digest": hex.EncodeToString(digest[:]), "accepted_at": now, "expires_at": g.view.DeadlineAt}
|
|
if err := emitEvent(ctx, tx, t.NamespaceID, workflow, g.view.Revision, "workflow", "workflow.signal.accepted", meta); err != nil {
|
|
return err
|
|
}
|
|
if _, err := appendAuditWithEvidence(ctx, tx, t, identity, id, "workflow_signal", "workflow.signal", in.Reason, in.EvidenceRef, nil, meta); err != nil {
|
|
return err
|
|
}
|
|
out.Signal, err = readWorkflowSignal(ctx, tx, t.NamespaceID, workflow, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
|
|
})
|
|
return out, err
|
|
}
|
|
func (s *WorkflowSignalStore) Get(ctx context.Context, t auth.Token, workflow, id core.ID) (signal.View, error) {
|
|
var out signal.View
|
|
err := s.transaction(ctx, t, signal.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
|
|
v, _, err := readWorkflowInstance(ctx, tx, t.NamespaceID, workflow, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
version, err := instanceVersion(ctx, tx, v)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, queue := range version.Analysis.Queues {
|
|
if !p.AllowsQueue(queue) {
|
|
return fault.NotFound
|
|
}
|
|
}
|
|
out, err = readWorkflowSignal(ctx, tx, t.NamespaceID, workflow, id)
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|