Workflow / State-Machine Engine

1import java.util.*;
2/** Local linear-workflow protocol. Claims are executed by an external trusted worker, not this engine. */
3public final class Solution {
4    private Solution() {}
5    public enum State { READY, RUNNING, WAITING, COMPENSATING, SUCCEEDED, FAILED, CANCELLED, MANUAL_REVIEW, COMPENSATION_FAILED }
6    public enum Outcome { SUCCESS, RETRYABLE, FATAL, UNKNOWN }
7    public record Step(String name,int maxAttempts,long retryDelay,boolean idempotent,boolean compensate){public Step{if(name==null||name.isBlank()||maxAttempts<1||maxAttempts>5||retryDelay<1)throw new IllegalArgumentException("step");}}
8    public record Definition(String version,List<Step> steps){public Definition{if(version==null||version.isBlank()||steps.isEmpty())throw new IllegalArgumentException("definition");steps=List.copyOf(steps);if(steps.stream().map(Step::name).distinct().count()!=steps.size())throw new IllegalArgumentException("duplicate step");}}
9    public record Claim(String instanceId,String definitionVersion,String input,String step,long token,boolean compensation,String externalKey,long leaseUntil){}
10    public record View(String id,String definitionVersion,State state,int completed,long due,long activeToken,boolean cancelRequested){}
11    private record Start(String version,String input,String instanceId){}
12    private static final class Instance {
13        final String id,input;final Definition definition;final List<Integer> completed=new ArrayList<>();final ArrayDeque<Integer> undo=new ArrayDeque<>();final Map<Long,Outcome> outcomes=new HashMap<>();
14        State state=State.READY;int index,attempts,undoAttempts;long nextToken,activeToken,leaseUntil,due;boolean cancel,compensating;
15        Instance(String id,String input,Definition definition){this.id=id;this.input=input;this.definition=definition;}
16    }
17    public static final class Engine {
18        private final long leaseDuration;
19        private final Map<String,Definition> definitions=new HashMap<>();
20        private final Map<String,Instance> instances=new HashMap<>();
21        private final Map<String,Start> starts=new HashMap<>();
22        private long sequence,time;
23        public Engine(long leaseDuration){if(leaseDuration<1)throw new IllegalArgumentException("lease");this.leaseDuration=leaseDuration;}
24        private void at(long now){if(now<time)throw new IllegalArgumentException("time");time=now;}
25        public synchronized void register(Definition definition){Objects.requireNonNull(definition);var old=definitions.get(definition.version());if(old!=null&&!old.equals(definition))throw new IllegalArgumentException("version is immutable");definitions.put(definition.version(),definition);}
26        private Instance instance(String id){var i=instances.get(id);if(i==null)throw new IllegalArgumentException("instance");return i;}
27        public synchronized View start(String version,String input,String key){if(input==null||key==null||key.isBlank())throw new IllegalArgumentException("start");var old=starts.get(key);if(old!=null){if(!old.version().equals(version)||!old.input().equals(input))throw new IllegalArgumentException("key reused");return view(old.instanceId());}var definition=definitions.get(version);if(definition==null)throw new IllegalArgumentException("definition");var i=new Instance("W"+(++sequence),input,definition);instances.put(i.id,i);starts.put(key,new Start(version,input,i.id));return view(i.id);}
28        private Step step(Instance i){return i.definition.steps().get(i.compensating?i.undo.peekFirst():i.index);}
29        public synchronized Claim claim(String id,long now){
30            at(now);var i=instance(id);if(!Set.of(State.READY,State.WAITING,State.COMPENSATING).contains(i.state)||now<i.due||i.activeToken!=0)throw new IllegalStateException("not claimable");
31            long until=Math.addExact(now,leaseDuration),token=Math.incrementExact(i.nextToken);i.nextToken=token;i.activeToken=token;i.leaseUntil=until;i.state=State.RUNNING;if(i.compensating)i.undoAttempts++;else i.attempts++;var step=step(i);return new Claim(i.id,i.definition.version(),i.input,step.name(),token,i.compensating,i.id+"/"+step.name()+"/"+(i.compensating?"undo":"work"),until);
32        }
33        private void beginUndo(Instance i){
34            i.activeToken=0;i.compensating=true;i.undo.clear();for(int n=i.completed.size()-1;n>=0;n--){int index=i.completed.get(n);if(i.definition.steps().get(index).compensate())i.undo.addLast(index);}i.undoAttempts=0;i.due=time;i.state=i.undo.isEmpty()?(i.cancel?State.CANCELLED:State.FAILED):State.COMPENSATING;
35        }
36        private void failed(Instance i,boolean retryable){
37            var step=step(i);int used=i.compensating?i.undoAttempts:i.attempts;
38            if(retryable&&step.idempotent()&&used<step.maxAttempts()&&!(!i.compensating&&i.cancel)){i.due=Math.addExact(time,step.retryDelay());i.state=i.compensating?State.COMPENSATING:State.WAITING;}
39            else if(i.compensating)i.state=State.COMPENSATION_FAILED;else beginUndo(i);
40        }
41        public synchronized View complete(String id,long token,Outcome outcome,long now){
42            Objects.requireNonNull(outcome);var i=instance(id);var old=i.outcomes.get(token);if(old!=null){if(old!=outcome)throw new IllegalStateException("contradictory completion");return view(id);}at(now);
43            if(i.state!=State.RUNNING||i.activeToken!=token)throw new IllegalStateException("stale completion");
44            if(now>=i.leaseUntil){Math.addExact(now,step(i).retryDelay());expire(i);throw new IllegalStateException("lease expired; reconcile external outcome");}
45            if(outcome==Outcome.UNKNOWN)return view(id);
46            // Prepare retry arithmetic before accepting an outcome that may need it.
47            if(outcome==Outcome.RETRYABLE)Math.addExact(now,step(i).retryDelay());
48            i.outcomes.put(token,outcome);i.activeToken=0;
49            if(outcome==Outcome.SUCCESS){
50                if(i.compensating){i.undo.removeFirst();i.undoAttempts=0;i.due=time;i.state=i.undo.isEmpty()?(i.cancel?State.CANCELLED:State.FAILED):State.COMPENSATING;}
51                else{i.completed.add(i.index);i.index++;i.attempts=0;i.due=time;if(i.cancel)beginUndo(i);else i.state=i.index==i.definition.steps().size()?State.SUCCEEDED:State.READY;}
52            }else failed(i,outcome==Outcome.RETRYABLE);return view(id);
53        }
54        private void expire(Instance i){var step=step(i);i.activeToken=0;int used=i.compensating?i.undoAttempts:i.attempts;if(!step.idempotent()||used>=step.maxAttempts()||(i.cancel&&!i.compensating)){i.state=State.MANUAL_REVIEW;return;}failed(i,true);}
55        public synchronized View timeout(String id,long token,long now){at(now);var i=instance(id);if(i.state!=State.RUNNING||i.activeToken!=token||now<i.leaseUntil)throw new IllegalStateException("not expired active attempt");Math.addExact(now,step(i).retryDelay());expire(i);return view(id);}
56        public synchronized View cancel(String id,long now){at(now);var i=instance(id);if(Set.of(State.SUCCEEDED,State.FAILED,State.CANCELLED,State.MANUAL_REVIEW,State.COMPENSATION_FAILED).contains(i.state))throw new IllegalStateException("terminal or review");i.cancel=true;if(i.state!=State.RUNNING&&!i.compensating)beginUndo(i);return view(id);}
57        public synchronized View retryCompensation(String id,long now){at(now);var i=instance(id);if(i.state!=State.COMPENSATION_FAILED)throw new IllegalStateException("no failed compensation");i.undoAttempts=0;i.due=now;i.state=State.COMPENSATING;return view(id);}
58        public synchronized View view(String id){var i=instance(id);return new View(i.id,i.definition.version(),i.state,i.completed.size(),i.due,i.activeToken,i.cancel);}
59    }
60}
61