IDictionary<int,Task> allTenantsWorkers;

 var tenantWorker = new TenantWoker(TenantId);
 await tenantWorker.Run();

Dictionary<int> allTenantsWorkers=...;


while(!cancelled)
{
   // новые сообщения могут появляться время от времени- они попадают в очередь через внешний, по отношению к этому, сервис
   var tenantId = _rabbitMQ.GetNextFromQueue(); 
   if( tenantid==null) continue;
   if( allTenantsWorkers.ContainsKey(tenantid.value)) continue; // избежать запуска дубликата тенант воркера

   var tenantWorker = WorkerFactory.CreateWorkerFor(tenantId);    

   // запустить воркер асинхронно и как либо дождаться его 
   var workerTask = Task.Run(async()=> await tenantWorker.Run()) // может выполняться часа 2-3 или более
// код ниже подходит для того что бы корректно сделать пост процессинг и удалить воркер из глобального списка??? или я туплю крепко 
            .ContunueWith(async (parentThread)=>
                {
                                  await parentThread;  // дождаться завершения родителя ???
                                  allTenantWorkers.Remove(tenantid);  // удалить из списка выполняющихся воркеров когда он завершит работу
                                 },TaskContinuationOptions.RunContinuationsAsynchronously);

   allTenantsWorkers.Add(tenantId,workerTask);   // добавить воркера и тенант который он обрабатывает в список для исключения 
}

   // Если по каким-то причинам workerTask у Вас отработает очень быстро и будет почти сразу Completed,
   // то ContinueWith может быть вызван также почти сразу.
   // Операция в нем - очистка словаря. 
   // Т.к. мы можем прийти к ней почти сразу, то получается, что удаление из словаря может быть выполнено до добавления. 
   // В случае реализации это сценария tenantId останется в словаре навечно.
   var workerTask = Task.Run(async()=> await tenantWorker.Run()) // может выполняться часа 2-3 или более
// код ниже подходит для того что бы корректно сделать пост процессинг и удалить воркер из глобального списка??? или я туплю крепко 
            .ContunueWith(async (parentThread)=>
                {
                                  await parentThread;  // дождаться завершения родителя ???
                                  allTenantWorkers.Remove(tenantid);  // удалить из списка выполняющихся воркеров когда он завершит работу
                                 },TaskContinuationOptions.RunContinuationsAsynchronously);

   allTenantsWorkers.Add(tenantId,workerTask);   // добавить воркера и тенант который он обрабатывает в список для исключения

public interface IQueueListener
{
    int? GetNextMessage();
}

public class Multiplexor
{
    private readonly IQueueListener m_Queue;
    private readonly Action<int, CancellationToken> m_Worker;
    private readonly HashSet<int> m_TenantsBeingProcessed; // Тенанты, обрабатываемые в данный момент

    public Multiplexor(IQueueListener queue, Action<int, CancellationToken> worker)
    {
        m_Queue = queue;
        m_Worker = worker;
    }

    public void Start(CancellationToken ct = default(CancellationToken))
    {
        while (!ct.IsCancellationRequested)
        {
            int? tenant = m_Queue.GetNextMessage();
            if (!tenant.HasValue)
            {
                // Тротлимся, если нет сообщений, чтобы не перегружать RMQ пустыми запросами (хотя, тут конечно все от имплементации QueueListener зависит)
                Task.Delay(TimeSpan.FromSeconds(30), ct).Wait(ct);
            };

            lock (m_TenantsBeingProcessed)
            {
                if (m_TenantsBeingProcessed.Contains(tenant.Value)) continue; // Тенант уже обрабатывается - игнорируем
                m_TenantsBeingProcessed.Add(tenant.Value);
            }

            Task.Factory.StartNew(() =>
            {
                try
                {
                    m_Worker(tenant.Value, ct);
                }
                finally
                {
                    //  Обработка закончена - уберем тенант из списка обрабатываемых
                    lock (m_TenantsBeingProcessed)
                    {
                        m_TenantsBeingProcessed.Remove(tenant.Value);
                    }
                }
            }, ct);
        }
    }

}

SL>   // Вы передаете в Task.Run async-функцию, она завершится на первом await, tenantWorker запустится, но ожидания не будет.
SL>   // Т.е. workerTask у Вас будет  сразу Completed, никаких 2-3 часов не пройдет.
SL>   // Дальше сразу вызовется ContinueWith, Ваш await в нем смысла не имеет: Вы ждете на уже завершившейся таске. 
SL>

Task.Run(async ()=> {await myType.Run();}).ContinueWith(...);