5805868292

pending completion

Build # 5805868292

Build Type

push

github

Committed by

lucaslorentz

Commit Message

Search by completed time

Run Details

12 of 12 new or added lines in 5 files covered. (100.0%)

2311 of 2794 relevant lines covered (82.71%)

145.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.36

/src/LLL.DurableTask.EFCore/OrchestrationDbContextExtensions.cs

using System;
using System.Linq;
using System.Threading.Tasks;
using DurableTask.Core;
using DurableTask.Core.Query;
using LLL.DurableTask.Core;
using LLL.DurableTask.EFCore.Entities;
using Microsoft.EntityFrameworkCore;

namespace LLL.DurableTask.EFCore
{
    public abstract class OrchestrationDbContextExtensions
    {
        public abstract Task Migrate(OrchestrationDbContext dbContext);

        public abstract Task WithinTransaction(OrchestrationDbContext dbContext, Func<Task> action);

        public async Task<T> WithinTransaction<T>(OrchestrationDbContext dbContext, Func<Task<T>> action)
        {
            T result = default;
            await WithinTransaction(dbContext, async () =>
            {
                result = await action();
            });
            return result;
        }

        public abstract Task<Instance> LockInstanceForUpdate(
            OrchestrationDbContext dbContext,
            string instanceId);

        public abstract Task<Instance> TryLockNextInstanceAsync(
            OrchestrationDbContext dbContext,
            TimeSpan lockTimeout);

        public abstract Task<Instance> TryLockNextInstanceAsync(
            OrchestrationDbContext dbContext,
            string[] queues,
            TimeSpan lockTimeout);

        public abstract Task<ActivityMessage> TryLockNextActivityMessageAsync(
            OrchestrationDbContext dbContext,
            TimeSpan lockTimeout);

        public abstract Task<ActivityMessage> TryLockNextActivityMessageAsync(
            OrchestrationDbContext dbContext,
            string[] queues,
            TimeSpan lockTimeout);

        public async Task PurgeOrchestrationHistoryAsync(
            OrchestrationDbContext dbContext,
            DateTime thresholdDateTimeUtc,
            OrchestrationStateTimeRangeFilterType timeRangeFilterType)
        {
            var query = timeRangeFilterType switch
            {
                OrchestrationStateTimeRangeFilterType.OrchestrationCreatedTimeFilter =>
                    dbContext.Executions.Where(e => e.CreatedTime < thresholdDateTimeUtc),
                OrchestrationStateTimeRangeFilterType.OrchestrationLastUpdatedTimeFilter =>
                    dbContext.Executions.Where(e => e.LastUpdatedTime < thresholdDateTimeUtc),
                OrchestrationStateTimeRangeFilterType.OrchestrationCompletedTimeFilter =>
                    dbContext.Executions.Where(e => e.CompletedTime < thresholdDateTimeUtc),
                _ => throw new NotImplementedException()
            };

            await ExecuteDeleteAsync(dbContext, query);
        }

        public async Task<int> PurgeInstanceHistoryAsync(
            OrchestrationDbContext dbContext,
            string instanceId)
        {
            var query = dbContext.Executions.Where(e => e.InstanceId == instanceId);

            return await ExecuteDeleteAsync(dbContext, query);
        }

        public async Task<int> PurgeInstanceHistoryAsync(OrchestrationDbContext dbContext, PurgeInstanceFilter filter)
        {
            var query = dbContext.Executions.Where(e => e.CreatedTime >= filter.CreatedTimeFrom
                && (filter.CreatedTimeTo == null || e.CreatedTime <= filter.CreatedTimeTo)
                && (filter.RuntimeStatus == null || filter.RuntimeStatus.Contains(e.Status)));

            return await ExecuteDeleteAsync(dbContext, query);
        }

        protected virtual Task<int> ExecuteDeleteAsync<T>(OrchestrationDbContext dbContext, IQueryable<T> query)
            where T : class
        {
            return query.ExecuteDeleteAsync();
        }

        public virtual IQueryable<Execution> CreateFilteredQueryable(
            OrchestrationDbContext dbContext,
            OrchestrationQuery query)
        {
            var extendedQuery = query as ExtendedOrchestrationQuery;

            var queryable = dbContext.Executions as IQueryable<Entities.Execution>;

            if (extendedQuery != null && !extendedQuery.IncludePreviousExecutions)
                queryable = queryable.Join(dbContext.Instances, x => x.ExecutionId, x => x.LastExecutionId, (x, y) => x);

            if (!string.IsNullOrEmpty(query.InstanceIdPrefix))
                queryable = queryable.Where(e => e.InstanceId.StartsWith(query.InstanceIdPrefix));

            if (query.CreatedTimeFrom != null)
                queryable = queryable.Where(e => e.CreatedTime >= query.CreatedTimeFrom);

            if (query.CreatedTimeTo != null)
                queryable = queryable.Where(e => e.CreatedTime <= query.CreatedTimeTo);

            if (query.RuntimeStatus != null && query.RuntimeStatus.Any())
                queryable = queryable.Where(e => query.RuntimeStatus.Contains(e.Status));

            if (extendedQuery != null)
            {
                if (!string.IsNullOrEmpty(extendedQuery.NamePrefix))
                    queryable = queryable.Where(e => e.Name.StartsWith(extendedQuery.NamePrefix));

                if (extendedQuery.CompletedTimeFrom != null)
                    queryable = queryable.Where(e => e.CompletedTime >= extendedQuery.CompletedTimeFrom);

                if (extendedQuery.CompletedTimeTo != null)
                    queryable = queryable.Where(e => e.CompletedTime <= extendedQuery.CompletedTimeTo);

                foreach (var kv in extendedQuery.Tags)
                {
                    var executionsWithTag = dbContext.Executions
                        .SelectMany(e => e.Tags, (e, t) => new { e, t })
                        .Where(e => e.t.Name == kv.Key && e.t.Value == kv.Value)
                        .Select(x => new { x.e.ExecutionId });

                    queryable = queryable.Join(
                        executionsWithTag,
                        x => x.ExecutionId,
                        x => x.ExecutionId,
                        (x, y) => x);
                }
            }

            var continuationToken = EFCoreContinuationToken.Parse(query.ContinuationToken);
            if (continuationToken != null)
            {
                queryable = queryable.Where(i =>
                    i.CreatedTime < continuationToken.CreatedTime ||
                    i.CreatedTime == continuationToken.CreatedTime && continuationToken.InstanceId.CompareTo(i.InstanceId) < 0);
            }

            return queryable;
        }
    }
}

1	using System;
2	using System.Linq;
3	using System.Threading.Tasks;
4	using DurableTask.Core;
5	using DurableTask.Core.Query;
6	using LLL.DurableTask.Core;
7	using LLL.DurableTask.EFCore.Entities;
8	using Microsoft.EntityFrameworkCore;
9
10	namespace LLL.DurableTask.EFCore
11	{
12	public abstract class OrchestrationDbContextExtensions
13	{
14	public abstract Task Migrate(OrchestrationDbContext dbContext);
15
16	public abstract Task WithinTransaction(OrchestrationDbContext dbContext, Func<Task> action);
17
18	public async Task<T> WithinTransaction<T>(OrchestrationDbContext dbContext, Func<Task<T>> action)
19	{
20	T result = default;	1,133✔
21	await WithinTransaction(dbContext, async () =>	1,133✔
22	{	1,133✔
23	result = await action();	2,266✔
24	});	1,133✔
25	return result;	1,133✔
26	}
27
28	public abstract Task<Instance> LockInstanceForUpdate(
29	OrchestrationDbContext dbContext,
30	string instanceId);
31
32	public abstract Task<Instance> TryLockNextInstanceAsync(
33	OrchestrationDbContext dbContext,
34	TimeSpan lockTimeout);
35
36	public abstract Task<Instance> TryLockNextInstanceAsync(
37	OrchestrationDbContext dbContext,
38	string[] queues,
39	TimeSpan lockTimeout);
40
41	public abstract Task<ActivityMessage> TryLockNextActivityMessageAsync(
42	OrchestrationDbContext dbContext,
43	TimeSpan lockTimeout);
44
45	public abstract Task<ActivityMessage> TryLockNextActivityMessageAsync(
46	OrchestrationDbContext dbContext,
47	string[] queues,
48	TimeSpan lockTimeout);
49
50	public async Task PurgeOrchestrationHistoryAsync(
51	OrchestrationDbContext dbContext,
52	DateTime thresholdDateTimeUtc,
53	OrchestrationStateTimeRangeFilterType timeRangeFilterType)
54	{
55	var query = timeRangeFilterType switch	24✔
56	{	24✔
57	OrchestrationStateTimeRangeFilterType.OrchestrationCreatedTimeFilter =>	24✔
58	dbContext.Executions.Where(e => e.CreatedTime < thresholdDateTimeUtc),	48✔
59	OrchestrationStateTimeRangeFilterType.OrchestrationLastUpdatedTimeFilter =>	24✔
60	dbContext.Executions.Where(e => e.LastUpdatedTime < thresholdDateTimeUtc),	24✔
61	OrchestrationStateTimeRangeFilterType.OrchestrationCompletedTimeFilter =>	24✔
62	dbContext.Executions.Where(e => e.CompletedTime < thresholdDateTimeUtc),	24✔
63	_ => throw new NotImplementedException()	24✔
64	};	24✔
65
66	await ExecuteDeleteAsync(dbContext, query);	24✔
67	}
68
69	public async Task<int> PurgeInstanceHistoryAsync(
70	OrchestrationDbContext dbContext,
71	string instanceId)
72	{
73	var query = dbContext.Executions.Where(e => e.InstanceId == instanceId);	6✔
74
75	return await ExecuteDeleteAsync(dbContext, query);	6✔
76	}
77
78	public async Task<int> PurgeInstanceHistoryAsync(OrchestrationDbContext dbContext, PurgeInstanceFilter filter)
79	{
80	var query = dbContext.Executions.Where(e => e.CreatedTime >= filter.CreatedTimeFrom	×
81	&& (filter.CreatedTimeTo == null \|\| e.CreatedTime <= filter.CreatedTimeTo)	×
82	&& (filter.RuntimeStatus == null \|\| filter.RuntimeStatus.Contains(e.Status)));	×
83
84	return await ExecuteDeleteAsync(dbContext, query);	×
85	}
86
87	protected virtual Task<int> ExecuteDeleteAsync<T>(OrchestrationDbContext dbContext, IQueryable<T> query)
88	where T : class
89	{
90	return query.ExecuteDeleteAsync();	21✔
91	}
92
93	public virtual IQueryable<Execution> CreateFilteredQueryable(
94	OrchestrationDbContext dbContext,
95	OrchestrationQuery query)
96	{
97	var extendedQuery = query as ExtendedOrchestrationQuery;	7✔
98
99	var queryable = dbContext.Executions as IQueryable<Entities.Execution>;	7✔
100
101	if (extendedQuery != null && !extendedQuery.IncludePreviousExecutions)	7✔
102	queryable = queryable.Join(dbContext.Instances, x => x.ExecutionId, x => x.LastExecutionId, (x, y) => x);	7✔
103
104	if (!string.IsNullOrEmpty(query.InstanceIdPrefix))	7✔
105	queryable = queryable.Where(e => e.InstanceId.StartsWith(query.InstanceIdPrefix));	×
106
107	if (query.CreatedTimeFrom != null)	7✔
108	queryable = queryable.Where(e => e.CreatedTime >= query.CreatedTimeFrom);	×
109
110	if (query.CreatedTimeTo != null)	7✔
111	queryable = queryable.Where(e => e.CreatedTime <= query.CreatedTimeTo);	×
112
113	if (query.RuntimeStatus != null && query.RuntimeStatus.Any())	7✔
114	queryable = queryable.Where(e => query.RuntimeStatus.Contains(e.Status));	×
115
116	if (extendedQuery != null)	7✔
117	{
118	if (!string.IsNullOrEmpty(extendedQuery.NamePrefix))	7✔
119	queryable = queryable.Where(e => e.Name.StartsWith(extendedQuery.NamePrefix));	×
120
121	if (extendedQuery.CompletedTimeFrom != null)	7✔
122	queryable = queryable.Where(e => e.CompletedTime >= extendedQuery.CompletedTimeFrom);	×
123
124	if (extendedQuery.CompletedTimeTo != null)	7✔
125	queryable = queryable.Where(e => e.CompletedTime <= extendedQuery.CompletedTimeTo);	×
126
127	foreach (var kv in extendedQuery.Tags)	34✔
128	{
129	var executionsWithTag = dbContext.Executions	10✔
130	.SelectMany(e => e.Tags, (e, t) => new { e, t })	10✔
131	.Where(e => e.t.Name == kv.Key && e.t.Value == kv.Value)	10✔
132	.Select(x => new { x.e.ExecutionId });	10✔
133
134	queryable = queryable.Join(	10✔
135	executionsWithTag,	10✔
136	x => x.ExecutionId,	10✔
137	x => x.ExecutionId,	10✔
138	(x, y) => x);	10✔
139	}
140	}
141
142	var continuationToken = EFCoreContinuationToken.Parse(query.ContinuationToken);	7✔
143	if (continuationToken != null)	7✔
144	{
145	queryable = queryable.Where(i =>	1✔
146	i.CreatedTime < continuationToken.CreatedTime \|\|	1✔
147	i.CreatedTime == continuationToken.CreatedTime && continuationToken.InstanceId.CompareTo(i.InstanceId) < 0);	1✔
148	}
149
150	return queryable;	7✔
151	}
152	}
153	}

lucaslorentz / durabletask-extensions / 5805868292

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous