1

I have n classes that extends from a State class, the purpose of the State class is to manage the state of the extended classes. For each class we need to save it in the database, remove it and select all the data in order to recover the classes to the same timestamp we stopped (in case of system failure or shutdown).

For example:

We have TransactionEntry that extends from State, each time we process some transaction, we insert it to the DB and when the processing is over, we remove it. In case our system suddenly shouted down, we need to recover all the Transaction objects (which did not processed successfully) from the DB to our Transaction list in the system in order to keep processing them.

Each Entry has a different way to recover (the way we select the data from the DB in order to add it to the list in the system) and some of them has the ability to update data in the DB.

For example:

FeedbackEntry can do all the things TransactionEntry can do but recover the data from the DB by State and update the current feedback state.
Feedbacks has state, like running or delayed so we want the ability to recover all the running (or delayed) feedbacks. Because feedback has state, and the state is changing sometimes (during the processing time), we need the ability to update the state in the DB.

So FeedbackEntry can do all the things TransactionEntry can do but the way we recover the data is different and because feedback has the ability to change state, we need to allow update query.

So far we discovered that we need to insert, remove and recover (by select) to our entries. The difference is in the way we recover them and the abilities that comes with each way.

Moreover, we want to add more abilities to some Entries. Some of them need the ability to Retry in case of Exception or in case something went wrong.

For example:

OverrideEntry can do all the things FeedbackEntry can do (which is recover by state, update state, insert, select, remove) and the ability to update retry count. OverrideEntry has some logic, that in case of Exception we must keep try to process for 3 / 4 times. So for each retry time, we have to update the retry counter in the DB.

As i understand, OverrideEntry needs the ability to insert, remove, select, recover by state, update state and update retry count.

How do i see the implementation ?

I see an abstract class called State:

public abstract class State<T extends StateEntry> implements Stateable<T> {
    // Some Connection object to the database

    @Override
    public void insert(T entry) {
    // insert entry to DB
    }

    @Override
    public void remove(T entry) {
    // delete entry from the DB
    }

    @Override
    public List<T> select(String query) {
        // select all the data by query
        // (this query is flexible in order to serve the recovery types)
        return null;
    }
}

Stateable interface:

public interface Stateable<T extends StateEntry> {
    void insert(T entry);
    void remove(T entry);
    List<T> select(String query);
}

They are exposing the base abilities of insert, select and remove.

Now, each entry will implement a Recoverable or RecoverableByState:

public interface Recoverable<T extends StateEntry> extends Stateable<T> {
    void recover(Addable<T> manager);
}

public interface RecoverableByState<T extends StateEntry> extends Stateable<T> {
    void recover(Addable<T> manager, int state);
    void updateState(T entry, int state);
}

Addable is an interface that allow us to add the recovered data to our list:

public interface Addable<T extends StateEntry> {
    void add(T entry);
}

All the logic on the entries is in Manager classes. Each manager manages the logic of the entry, for example:

public class TransactionManager implements Addable<TransactionEntry> {

    private ArrayList<TransactionEntry> transaction;
    void checkTransaction(TransactionEntry entry){
        // ... logic

        // In other entries, their manager can change things on the entry 
        // that we must record this change in the DB, like state (from delayed to running) 
        // or update retry count or something like that, so for some entries
        // we need the ability to update the fields we need:
        // overrideState.updateRetryCount(overrideEntry, 3); // will update the retry count of overrideEntry to 3
        // or overrideState.recover(new OverrideManager(), State.running)
    }

    void closeTransaction(TransactionEntry entry) {
        // Remove from state
        // state.remove(entry)
        transaction.remove(entry);
    }

    @Override
    public void add(TransactionEntry entry) {
        // Insert entry to state
        // state.insert(entry)
        transaction.add(entry);
    }
}

Now, for each entry that needs to implement more abilities, we will create and implement an interface like:

public interface Retryable<T extends StateEntry> {
    Connection postgresConnection();

    default void updateRetryCount(T entry, int retryCount) {
        String query = "update entries_state set retry = {0} where id = {1}";
        {0} -> retryCount
        {1} -> entry.getID()
        this.postgresConnection.execute(query);
    }
}

Example of entries:

A base entry, contains all the common fields of the entries:

public abstract class StateEntry {
    // Will have all the common fields of the entries
    // like guid, record_creation_time, entry_name...
}

On TransactionEntry we need simple recover without special abilities

public class TransactionEntry extends StateEntry {
    @JsonProperty("TransactionID")
    private String transactionId;

    @JsonProperty("NumOfEntitiesOnTransaction")
    private int entitiesOnTransaction;

    @JsonProperty("CreationTime")
    private Instant creationTime;

    // Ctor
}

On FeedbackEntry we need to recover by state and also update the state

public class FeedbackEntry extends StateEntry {

    @JsonProperty("Host")
    private String host;

    @JsonProperty("Feedback")
    private JsonNode feedback;

    @JsonProperty("SendingTime")
    private Instant sendingTime;

    @JsonProperty("StateType")
    private StateType stateType;

    // Ctor
}

On OverrideEntry we need to recover by state and

public class OverrideEntry extends StateEntry {
    @JsonProperty("Message")
    private String message;

    @JsonProperty("StartProcessingTime")
    private Instant startProcessingTime;

    @JsonProperty("RetryCount")
    private int retryCount;

    @JsonProperty("StateType")
    private StateType stateType;

    // Ctor

}

How it looks in the Database ?

I see a single table, called entries_state which each field in StateEntry (the base class that all the entries extends from) is a column and one more column of type Json that contains the entry itself after serialization from Object to Json.

My question

Does it sound and looks good for you ?.

I want to be able to create states easily as:

Stateable<TransactionEntry> transactionState = StateFactory.createInstance(StateType.Recover);

Stateable<FeedbackEntry> feedbackState = StateFactory.createInstance(StateType.RecoveryByState);

Stateable<OverrideEntry> overrideState = StateFactory.createInstance(StateType.RecoveryByState, StateType.Retry);

I mean i want to create some Factory that will create the wanted combination of state abilities as i pass. I do not think this is possible in the way i want to implement so i will glad to hear your opinions, maybe all the way i think is wrong. I'm thinking about a solution for 3 days and can't get a good one.

Thank you all for your time and your helping.

Ben.

0

The problem of going down the state-full object path, is the problem of synchronisation. You would have to write custom code to detect degenerate states in the system and correct them. This is possible to do, after all ACID databases are written in this way. But it is by no means easy.

Based on what you are describing, I feel that you should consider workflows and tasks.

Generally a task or workflow will manipulate many objects within your system in the course of achieving its goal, but those manipulations occur only when the task completes, or the workflow moves to the next stage.

Tasks

A Task is usually binary (though it doesn't have to be defined that way), with two states of interest: Ready, and Complete. When the task is picked up:

  • an entry could be added identifying the attempt.
  • A transaction with the db would be started,
  • any operations on state performed,
  • the task record itself updated to being complete,
  • and any new spin off task created.
  • Then the transaction would be committed.

If the task rolled-back, the log would reveal that there was an attempt, but no other state would have changed.

If the task committed, the log would reveal that there was an attempt, and the task would be complete. The state change would also have occurred, along with any follow on tasks being recorded.

Something like:

enum Status { Success, PartialSuccess, Warning, Failed }
enum Semantic { .... }
interface ReportItem
{
     Status status { get; }
     Semantic semantic { get; }
     string message { get; }
}
interface Report
{
     bool Success { get; }
     IList<ReportItem> items { get; }
}

interface Task
{
     string Label {get;}
     Workflow { get;}

     Report run(DabaseConnector db); //<option 1
     void run(Report report, DabaseConnector db); //<option 2
}
  • The Report is a description about how the task ran, you could either pass it in as an argument (option 2) or exepect it to be returned (option 1).
  • Semantic is a special purpose enumeration to allow your task to report known issues back up to someone/thing that can do something to fix it. This is useful with workflows that can handle errors cases based on this semantic in a more intelligent manner.
  • The Label is something human readable so that support can determine what the task was. It also helps for logging purposes.
  • Workflow I think is self-evident. If its associated with one it is set to it, otherwise its null.

    void TaskQueue(KillSignal signal, Queue queue, DatabaseConnection db)
    {
       while (!signal)
       {
           Task task;
           while (task = queue.next())
           {
               Report report = null;
               try
               {
                   using (var transaction = db.beginTransaction())
                   {
                       report = task.run(transaction);
                       if (task.workflow != null)
                           report = task.workflow.next(task, report);
    
                       if (report.success)
                           transaction.commit();
                       else
                           transaction.rollback();
                   }
               }
               catch(Exception ex)
               {
                   enrichReport(ref report, ex);
               }
               queue.complete(task, report);
           }
       }
    }
    

Generally there will be a queue. The queue could be a third-party offering. Or simply a transactional table. The only requirement is that queue.next() picks a task and somehow reserves it for a period of time. Maybe the queue is single threaded, or there is some other synchronisation mechanism at play.

The a transaction is then created, and the task is executed within the transaction. Depending on how your database works you may need to rejig the pattern of trys/using etc... The goal is to ensure that the transaction only commits if the job returns success, and rolls back every other time.

Regardless of how the task transaction goes, the queue still gets to update the task record with the report. That can cause a reschedule, or simply halt the job for manual intervention.

If the task is part of a worklow, it is consulted. This gives the workflow to imbue the task failure with more diagnostics, or allow the workflow to automatically correct the issue, or allow the workflow to generate new tasks.

Workflows

A Workflow can be seen as the chains of these tasks. Perhaps even be pointed to by the task both to highlight why the task is being run, but also to allow the task to say, yep done the job that corresponds to this point in the workflow, what other tasks need to be queue now? This allows the separation of choreography from implementation.

Roughly a workflow looks like.

interface Workflow
{
    String label {get; }

    Task start(...Workkflow Arguments...);

    Report next(Task task, Report report);
}

start(...) is called passing in any arguments relevant for the workflow. It returns a task that represents the first step of the workflow. If you so desire, the task could be sent directly to the queue instead, it will depend on how you've architectured your persistence layer.

The next(...) is task with deciding what the workflow will do next. It might improve the report generated by the task, it might generate new tasks, it might do nothing, or update its own state.

A good way to implement workflows are as Finite State Machines.

More thoughts on tasks

Just extending Tasks, think of them as a mini-process:

  • New - the task is added and has yet to run
  • Running - the task is currently running
  • Failed - the task could not bring about the desired state change, and cannot do so.
  • Success - the task finished
  • Blocked (the task needs a resource in order to complete, its awaiting its availability)

It would make sense to treat running as orthogonal to the other states. Largely because should a crash occur it makes rollback simple. Update all tasks with the running flag to not running. Alternately you could time bound the tasks so that a task cannot run more than say 20 minutes before being terminated, which conveniently also allows a crashed server to resume crashed task say 25 minutes after the timestamp...

  • Thank you for your answer but can you add some examples?. I did not undersand your idea completly. – bnsd55 Feb 28 at 6:21
  • I've expounded a bit further, let me know if you've a specific issue you'd like me to delve into further. – Kain0_0 Mar 3 at 23:32

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.