social-native / kafka-sagas Goto Github PK

I would like to propose some additional syntax to control the behavior of an action channel. Initially, I would like to support two new options: transactionVisibility and deserializer.

• transactionVisibility can be one of three kinds:
  • {kind: 'local_transactions'} (default)
    • Would only buffer actions from the same transaction as the initiating action
  • {kind: 'transaction_id', transaction_id: string}
    • Would only buffer actions that match the given transaction id
  • {kind: 'global'}
    • Would buffer all actions from a topic
    • Since the default buffer is unbounded, a different buffer would need to be provided to give a boundedness to the action channel.
• deserializer is an option that can replace the default message deserializer, which expects a JSON of shape { transaction_id, headers, payload }. This deseralizer would still be required to have an output interface of IAction.

The syntax extension I am proposing is as follows:

class AvroDeserializer extends ActionChannelDeserializer {
}

const deserializer = new AvroDeserializer();

function* saga(action, context) {
    const channel = context.effects.actionChannel('my_topic', predicateFunction, {
        transactionVisibility: {kind: 'global'},
        deserializer
    });
}

We outlined the following uses for this package:

• CLI: Create helm deployment for worker per service.
  - Dedicated tiny little sweet baby jesus pod per worker file that's listening and crunching using the function.

Worker pod's execution flow

• Receive a job from kafka per-job topic
  { job_id: asdf1234 }

• Jobs table for just ids

• Per-job request table for work to be done { job_id }

• Per-job response table for work completed { job_id }

• Query mysql for per-job metadata

  • Single worker pod per job that's listening to one topic.
    e.g.
    {topic: KAFKA.CREATE_CAMPAIGN, action: (request, response, query, mutate) => {}}

• Call action function in src/background_workers/*

  • This should be abstracted such that this part can be either done on-pod-premises or eventually elsewhere

• Notify per-job table work was done

In the Saga pattern, compensation is a key aspect of the workflow. This RFC proposes a new effect for us to register compensation to be handled when a saga throws, which will not be subject to the overall timeout of the saga unless otherwise specified.

/** Eagerly send work to some other saga for compensation to be handled. */
type DistributedCompensator<Payload extends DefaultPayload> = {
    topic: string;
    payload: Payload;
}

/** Compensate with your own function, async function, or workflow (i.e. callable saga). */
type LocalCompensator<Payload extends DefaultPayload> = {
    handler: (payload: Payload) =>  void | Promise<void> | Generator<IEffectDescription, void, any>;
    payload: Payload;
}

type CompensationPlan<Payload> = DistributedCompensator<Payload> | LocalCompensator<Payload>;

type DeleteCampaignPayload = {campaign_id: string | number};

function* saga ({payload}, {context: {effects: {callFn, addCompensation}}}) {
    const campaignId = yield callFn(createCampaign, [{name: 'test_campaign'}]);

    yield addCompensation<DeleteCampaignPayload>({
        topic: topics.DELETE_CAMPAIGN,
        payload: {campaign_id: campaignId}
    });
}