Session.ForTenant and MultiTenantedWithSingleServer behavior

[AlexZeitler]

I have an event ThreadStarted:

public record ThreadStarted(string SenderSubscriptionId, string ReceiverSubscriptionId, string Topic, DateTimeOffset On, string By, string Message);

And a Projection Thread:

public record Thread
{
  private List<ThreadMessage> messages;
  public Guid Id { get; init; }
  public DateTimeOffset StartedOn { get; }
  public string StartedBy { get; }
  public string Topic { get; }

  public IReadOnlyCollection<ThreadMessage> Messages
  {
    get { return messages.AsReadOnly(); }
    private set => messages = value.ToList();
  }

  private Thread()
  {
  }

  private Thread(DateTimeOffset startedOn, string startedBy, string firstMessage, string topic)
  {
    StartedOn = startedOn;
    StartedBy = startedBy;
    Topic = topic;
    messages = new List<ThreadMessage> { new(startedOn, startedBy, firstMessage) };
  }

  public static Thread Create(ThreadStarted started)
  {
    return new Thread(started.On, started.By, started.Message, started.Topic);
  }
}

I have tested these in different scenarios with an DocumentStore setup with MultiTenantedWithSingleServer:

private static DocumentStore GetDocumentStore()
  {
    return DocumentStore.For(options =>
    {
      options
        .MultiTenantedWithSingleServer(GetConnectionString());
      options.AutoCreateSchemaObjects = AutoCreate.All;
      options.Projections.SelfAggregate<Thread>(ProjectionLifecycle.Inline);
      options.UseDefaultSerialization(
        EnumStorage.AsString,
        nonPublicMembersStorage: NonPublicMembersStorage.All
      );
    });
  }

First, the succeeding test where every tenant has its own session:

  [Fact]
  public async Task SeparateSessionsShouldCreateThreadsInBothTenants()
  {
    var store = GetDocumentStore();

    var topic = Guid.NewGuid().ToString();
    var @event = new ThreadStarted("ten1", "ten2", topic, DateTimeOffset.Now, "Jane Doe", "Hello World!");

    await using var writeSession1 = store.LightweightSession("ten1");
    await using var writeSession2 = store.LightweightSession("ten2");
    var streamId1 = Guid.NewGuid();
    var streamId2 = Guid.NewGuid();
    writeSession1.Events.StartStream(streamId1, @event);
    writeSession2.Events.StartStream(streamId2, @event);
    await writeSession1.SaveChangesAsync();
    await writeSession2.SaveChangesAsync();

    await using var readSessionTenant1 = store.LightweightSession("ten1");
    await using var readSessionTenant2 = store.LightweightSession("ten2");

    var thread1 = readSessionTenant1.Load<Thread>(streamId1);
    var thread2 = readSessionTenant2.Load<Thread>(streamId2);

    Assert.NotNull(thread1);
    Assert.NotNull(thread2);
  }

Next, I have failing test where I have a single session for both events:

  [Fact]
  public async Task SingleSessionWithFirstTenantShouldCreateThreadsInBothTenants()
  {
    var store = GetDocumentStore();

    var topic = Guid.NewGuid().ToString();
    var @event = new ThreadStarted("ten1", "ten2", topic, DateTimeOffset.Now, "Jane Doe", "Hello World!");

    await using var session = store.LightweightSession("ten1");
    var streamId1 = Guid.NewGuid();
    var streamId2 = Guid.NewGuid();
    session.ForTenant("ten1").Events.StartStream(streamId1, @event);
    session.ForTenant("ten2").Events.StartStream(streamId2, @event);
    await session.SaveChangesAsync();

    await using var readSessionTenant1 = store.LightweightSession("ten1");
    await using var readSessionTenant2 = store.LightweightSession("ten2");

    var thread1 = readSessionTenant1.Load<Thread>(streamId1);
    var thread2 = readSessionTenant2.Load<Thread>(streamId2);

    Assert.NotNull(thread1);
    Assert.NotNull(thread2);
  }

The database for ten2 has only one table: mt_doc_thread but it is empty. The ten1 database has the mt_doc_thread and mt_events tables having both events and both projections.

The next failing test is similar to the second but its session is initialized with a different tenant id:

  [Fact]
  public async Task SessionWithFirstRandomShouldCreateThreadsInBothTenants()
  {
    var store = GetDocumentStore();

    var topic = Guid.NewGuid().ToString();
    var @event = new ThreadStarted("ten1", "ten2", topic, DateTimeOffset.Now, "Jane Doe", "Hello World!");

    await using var session = store.LightweightSession("some-tenant");
    var streamId1 = Guid.NewGuid();
    var streamId2 = Guid.NewGuid();
    session.ForTenant("ten1").Events.StartStream(streamId1, @event);
    session.ForTenant("ten2").Events.StartStream(streamId2, @event);
    await session.SaveChangesAsync();

    await using var readSessionTenant1 = store.LightweightSession("ten1");
    await using var readSessionTenant2 = store.LightweightSession("ten2");

    var thread1 = readSessionTenant1.Load<Thread>(streamId1);
    var thread2 = readSessionTenant2.Load<Thread>(streamId1);

    Assert.NotNull(thread1);
    Assert.NotNull(thread2);
  }

The behavior is quite similar to the second test but now I have three databases some-tenant, ten1 and ten2, where some-tenant contains all events and all projections.

The repository can be found here.

[jeremydmiller]

I'm finally looking at this today

[AlexZeitler]

Just debugged this one. Did I get it right that session.TenantId is not holding the correct value ten2 here but ten1?

https://github.com/JasperFx/marten/blob/64598221e07050a8d1d82ef855011994e094260a/src/Marten/Events/EventGraph.Processing.cs#L126

[AlexZeitler]

I think @event also has the wrong TenantId but stream has the right TenantId for the operation.

[jeremydmiller]

This is never going to work to span tenants from multiple databases. We actually need a little bit of validation here to say "nope, this tenant is not part of the current database" There's no way to span a transaction across multiple databases.

[AlexZeitler]

Makes sense.