Trains & Junctions

Junctions

Junctions are the points along a train's route. Each one does one thing:

public class ValidateEmailJunction(IUserRepository UserRepository) : Junction<CreateUserRequest, Unit>
{
    public override async Task<Unit> Run(CreateUserRequest input)
    {
        var existingUser = await UserRepository.GetByEmailAsync(input.Email);
        if (existingUser != null)
            throw new ValidationException($"Email {input.Email} already exists");
 
        return Unit.Default;
    }
}
 
public class CreateUserJunction(IUserRepository UserRepository) : Junction<CreateUserRequest, User>
{
    public override async Task<User> Run(CreateUserRequest input)
    {
        var user = new User
        {
            Email = input.Email,
            FullName = $"{input.FirstName} {input.LastName}",
            CreatedAt = DateTime.UtcNow
        };
 
        return await UserRepository.CreateAsync(user);
    }
}

Junctions use constructor injection for dependencies. When a junction throws, the train switches to the left track and returns the exception.

CancellationToken in Junctions

Every junction has a CancellationToken property that is set automatically by the train before Run() is called. Use it to pass cancellation to async operations:

public class FetchUserJunction(IHttpClientFactory httpFactory) : Junction<UserId, UserProfile>
{
    public override async Task<UserProfile> Run(UserId input)
    {
        var client = httpFactory.CreateClient();
        var response = await client.GetAsync($"/users/{input.Value}", CancellationToken);
        response.EnsureSuccessStatusCode();
        return await response.Content.ReadFromJsonAsync<UserProfile>(CancellationToken);
    }
}

The token comes from the caller: train.Run(input, cancellationToken). If no token is provided, CancellationToken defaults to CancellationToken.None. Before each junction executes, cancellation is checked. If the token is already cancelled, the junction is skipped and OperationCanceledException propagates.

Full details: Cancellation Tokens

EffectJunction vs Junction

Trax has two junction base classes:

Junction<TIn, TOut> is the base class. Handles input/output and railway error propagation. No metadata, no lifecycle hooks. Use this for lightweight junctions or when running inside a plain Train.

EffectJunction<TIn, TOut> extends Junction with per-junction metadata tracking. When run inside a ServiceTrain, it records a JunctionMetadata entry with the junction's name, input/output types, start/end times, and railway state. Junction effect providers (like AddJunctionLogger) hook into EffectJunction's lifecycle and fire before and after each junction executes.

// Base junction, no metadata tracking
public class ValidateEmailJunction(IUserRepository repo) : Junction<CreateUserRequest, Unit>
{
    public override async Task<Unit> Run(CreateUserRequest input)
    {
        if (await repo.GetByEmailAsync(input.Email) != null)
            throw new ValidationException("Email already exists");
        return Unit.Default;
    }
}
 
// Effect junction, tracked by junction effect providers
public class ValidateEmailJunction(IUserRepository repo) : EffectJunction<CreateUserRequest, Unit>
{
    public override async Task<Unit> Run(CreateUserRequest input)
    {
        if (await repo.GetByEmailAsync(input.Email) != null)
            throw new ValidationException("Email already exists");
        return Unit.Default;
    }
}

The implementation is identical. Just swap the base class. EffectJunction only adds metadata when running inside a ServiceTrain. If you use EffectJunction inside a plain Train, it throws at runtime.

Use EffectJunction when you want junction-level observability (timing, logging via AddJunctionLogger). Use Junction when you don't need it.

Dependency Injection in Junctions

Junctions use standard constructor injection for their dependencies. Do not use the [Inject] attribute. That's used internally by the ServiceTrain base class for its own framework-level services.

// Don't use [Inject] in your junctions
public class MyJunction : Junction<Input, Output>
{
    [Inject]
    public IMyService MyService { get; set; }
}
 
// Use constructor injection
public class MyJunction(IMyService MyService) : Junction<Input, Output>
{
    public override async Task<Output> Run(Input input)
    {
        var result = await MyService.DoSomethingAsync(input);
        return result;
    }
}

Train Structure

As your application grows, you'll want a consistent way to organize trains. Group each train with its input, interface, and junctions in a single folder:

Trains/
├── CreateUser/
│   ├── CreateUserRequest.cs        # Input model
│   ├── ICreateUserTrain.cs         # Interface
│   ├── CreateUserTrain.cs          # Implementation
│   └── Junctions/
│       ├── ValidateEmailJunction.cs
│       ├── CreateUserInDatabaseJunction.cs
│       └── SendWelcomeEmailJunction.cs
│
├── ProcessOrder/
│   ├── ProcessOrderRequest.cs
│   ├── IProcessOrderTrain.cs
│   ├── ProcessOrderTrain.cs
│   └── Junctions/
│       ├── ValidateOrderJunction.cs
│       ├── ChargePaymentJunction.cs
│       └── CreateShipmentJunction.cs

The Input Model

Each train gets its own request type. This is required by the TrainBus because input types must be unique across your application:

namespace YourApp.Trains.CreateUser;
 
public record CreateUserRequest
{
    public required string Email { get; init; }
    public required string FirstName { get; init; }
    public required string LastName { get; init; }
}

The Interface

Define an interface for DI resolution and testing:

namespace YourApp.Trains.CreateUser;
 
public interface ICreateUserTrain : IServiceTrain<CreateUserRequest, User>;

The Junctions Folder

Junctions go in a Junctions/ subfolder. Mark them internal since they're implementation details of this train:

namespace YourApp.Trains.CreateUser.Junctions;
 
internal class ValidateEmailJunction(IUserRepository userRepository) : Junction<CreateUserRequest, Unit>
{
    public override async Task<Unit> Run(CreateUserRequest input)
    {
        var existing = await userRepository.GetByEmailAsync(input.Email);
        if (existing != null)
            throw new ValidationException($"Email {input.Email} already exists");
 
        return Unit.Default;
    }
}

Using internal keeps your public API surface clean. External code interacts with ICreateUserTrain, not individual junctions.

When to Share Junctions

Sometimes multiple trains need the same validation or transformation. Resist the urge to share junctions too early. Duplication is often cheaper than the wrong abstraction.

When you do need to share, create a Shared/ folder at the Trains/ level:

Trains/
├── Shared/
│   └── Junctions/
│       └── ValidateEmailFormatJunction.cs
├── CreateUser/
│   └── ...
├── UpdateUser/
│   └── ...

Shared junctions should be truly generic. If you find yourself adding conditionals to handle different trains, that's a sign the junction should be duplicated and specialized instead.

Defining a Train

Override Junctions() to define the route, the sequence of junctions the train passes through:

public class CreateUserTrain : ServiceTrain<CreateUserRequest, User>, ICreateUserTrain
{
    protected override Task<Either<Exception, User>> Junctions() =>
        Chain<ValidateEmailJunction>()
            .Chain<CreateUserJunction>().Resolve();
}

Junctions() returns TReturn directly. There is no Either, no async Task, no Activate, no Resolve. The framework handles all of that. Chain methods (Chain, ShortCircuit, Extract, AddServices) are available as protected methods on the train itself.

When to use RunInternal

For advanced cases (custom logic before/after the chain, manual Either construction, or passing extra objects into Memory), override RunInternal instead:

public class CreateUserTrain : ServiceTrain<CreateUserRequest, User>, ICreateUserTrain
{
    protected override async Task<Either<Exception, User>> RunInternal(CreateUserRequest input) =>
        Activate(input)
            .Chain<ValidateEmailJunction>()
            .Chain<CreateUserJunction>()
            .Resolve();
}

RunInternal gives you full control: Activate(input) seeds Memory, .Chain<T>() adds junctions, and .Resolve() extracts the result as Either<Exception, TReturn>. Use it when Junctions() isn't expressive enough.

Train Lifecycle Hooks

ServiceTrain provides protected virtual methods you can override to react to your train's own lifecycle events, with no global hook registration needed:

public class CreateUserTrain(ISlackClient slack)
    : ServiceTrain<CreateUserRequest, User>, ICreateUserTrain
{
    protected override User Junctions() =>
        Chain<ValidateEmailJunction>()
            .Chain<CreateUserJunction>();
 
    protected override async Task OnFailed(
        Metadata metadata, Exception exception, CancellationToken ct)
    {
        await slack.PostAsync($"User creation failed: {exception.Message}", ct);
    }
}

Available overrides: OnStarted, OnCompleted, OnFailed, OnCancelled. All default to no-op. Exceptions in overrides are caught and logged and never cause the train to fail. Each hook receives the metadata populated with everything known at that point: OnStarted sees the typed input (via TrainInput or metadata.GetInput<T>()); OnCompleted additionally sees the output; OnFailed/OnCancelled see the failure state.

These work alongside global lifecycle hooks. Global hooks fire first, then per-train overrides.

OnQueue (enqueue-time hook)

OnQueue is a fifth override that fires at a different moment from the others: synchronously inside the mediator's QUEUE path (ITrainExecutionService.QueueAsync), before the work queue row is inserted. The other four fire when the train runs; OnQueue fires when a QUEUE mutation is accepted, before the train is scheduled. It does not fire on the synchronous RUN path, and it does not fire again when the scheduler later runs the train.

Use it to perform a side-effect the instant a mutation is accepted, instead of waiting for the deferred run. A common case is an optimistic write: persist a provisional row so the change is visible immediately, then let the real run reconcile it.

public class ProcessMatchResultTrain
    : ServiceTrain<ProcessMatchResultInput, ProcessMatchResultOutput>, IProcessMatchResultTrain
{
    [Inject]
    public IDbContextFactory<GameDbContext>? GameDbFactory { get; set; }
 
    protected override async Task OnQueue(Metadata metadata, CancellationToken ct)
    {
        var input = metadata.GetInput<ProcessMatchResultInput>();
        if (input is null || GameDbFactory is null)
            return;
 
        await using var db = await GameDbFactory.CreateDbContextAsync(ct);
        db.Matches.Add(new MatchRecord { MatchId = input.MatchId, Region = input.Region });
        await db.SaveChangesAsync(ct);
    }
}

OnQueue differs from the other four hooks in three ways:

• Exceptions propagate. A throw is not swallowed: it aborts the enqueue and no work queue row is written. Use it only for work that must succeed for the mutation to be accepted.
• The train is not initialized. this.Metadata and TrainInput are unavailable. Read everything from the passed metadata: the input via metadata.GetInput<T>(), and metadata.ExternalId to correlate with the eventual run, which executes under the same ExternalId. Id, ManifestId, and ScheduledTime are unset because no run exists yet.
• It must be idempotent. The deferred run re-executes the full Junctions() chain, so any effect the chain also performs will happen again. Write OnQueue so running it plus the chain is safe.

Property dependencies marked [Inject] (like GameDbFactory above) are populated before OnQueue is called, the same as during a normal run. Trains that do not override OnQueue skip resolution entirely, so the enqueue path is unaffected.

SDK Reference

> Junctions | Activate | Chain | Resolve