DataAdapter issue in Scheduler

Question

I'm using a direct PostgreSQL database connection for the Scheduler's DataAdapter due to API performance issues. However, when using the DataAdapter.ReadAsync method, I encounter an Npgsql.NpgsqlOperationInProgressException: 'A command is already in progress.' It is observed in both Entity Framework (EF) and ADO.NET DataReader (registered to dependency injection transient). I noticed the issue arises because the Scheduler executes the same query multiple times simultaneously.Switching to the synchronous DataAdapter.Read resolves the concurrency error but results in a SocketException: An existing connection was forcibly closed by the remote host or an NpgsqlException: 'Exception while reading from stream' after idling.Is there a more effective solution to handle this scenario?public class ScheduleDataAdapter(ScheduleDataService service) : DataAdaptor{ //public override async Task

Swathi Ravi · Answer

Hi Yongkee Cho,

Thank you for reaching out to us.

The issue you are experiencing is related to the use of the foreach loop within the GetAsync method when calling db.GetSchedulesAsync. This results in simultaneous database calls, leading to the Npgsql.NpgsqlOperationInProgressException error you encountered.

In our Scheduler, the ReadAsync and Read methods are designed to trigger only once during the data load process. They will not invoke multiple database calls simultaneously, which might help avoid concurrency-related issues.

Regards,

Swathi

Yongkee Cho · Answer

You have to use foreach at least once to get all the results.

I used the foreach later for IAsyncEnumerable<Schedule> in the example above.

My other example, that returning Task<IList<Schdule>> from ADO.NET DataReader also causes the same issue.


    public static async IAsyncEnumerable<Schedule> GetSchedulesAsync(this IMyDb db, DateTime startTime, DateTime endTime, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        if (db == null) throw new ArgumentNullException(nameof(db));


        const string commandText = @"select b.id, b.subject, b.start_time, b.end_time, b.""location"", b.description, b.all_day, b.read_only, b.equipment_id, e.asset_type, e.manufacturer, e.model, b.recurrence_id, b.recurrence_rule, b.recurrence_exception, b.reserved_by, b.""archived$""
from equipment_booking1$raw b
join shared_equipment$raw e on e.id = b.equipment_id
where b.""archived$"" = false and b.start_time >= :startTime and b.end_time <= :endTime";


        var inParameters = new Dictionary<string, object>
        {
            { "startTime", startTime },
            { "endTime", endTime },
        };
        var reader = await db.ExecuteReaderAsync(commandText, inParameters, CommandBehavior.Default, cancellationToken);
        while (await reader.ReadAsync(cancellationToken))
            yield return reader.ReadSchedule();
        reader.Close();
    }


    public static async Task<IList<Schedule>> GetSchedulesAsync(this IMyDb db, DateTime startTime, DateTime endTime, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        if (db == null) throw new ArgumentNullException(nameof(db));


        const string commandText = @"select b.id, b.subject, b.start_time, b.end_time, b.""location"", b.description, b.all_day, b.read_only, b.equipment_id, e.asset_type, e.manufacturer, e.model, b.recurrence_id, b.recurrence_rule, b.recurrence_exception, b.reserved_by, b.""archived$""
from equipment_booking1$raw b
join shared_equipment$raw e on e.id = b.equipment_id
where b.""archived$"" = false and b.start_time >= :startTime and b.end_time <= :endTime";


        var inParameters = new Dictionary<string, object>
        {
            { "startTime", startTime },
            { "endTime", endTime },
        };
        var reader = await db.ExecuteReaderAsync(commandText, inParameters, CommandBehavior.Default, cancellationToken);
        var schedules = new List<Schedule>();
        while (await reader.ReadAsync(cancellationToken))
            schedules.Add(reader.ReadSchedule());
        reader.Close();


        return schedules;
    }

Yongkee Cho · Answer

Is there any chance that the query in DataAdapter.ReadAsync is called multiple times in my Shedule component?

Swathi Ravi · Answer

Yongkee Cho, You have to use foreach at least once to get all the results. Instead of using `foreach` to collect the data, you can directly call the GetSchedulesAsync method within the GetAsync method. public async Task> GetAsync(DateTime start, DateTime end) { // db.GetSchedulesAsync returns an asynchronous enumerable (IAsyncEnumerable) return await db.GetSchedulesAsync(start, end).ToListAsync(); } Is there any chance that the query in DataAdapter.ReadAsync is called multiple times in my Schedule component? In your scheduler, the foreach loop is causing multiple triggers.

Yongkee Cho · Answer

It seems you are using Reactive System.Linq.Async.

The implementation inside of their ToListAsync() also iterates through the list. https://github.com/dotnet/reactive/blob/fefe75993de542bc8dc6e01287fce0956a6770ab/Ix.NET/Source/System.Linq.Async/System/Linq/Operators/ToList.cs#L22

Swathi Ravi · Answer

Yongkee Cho,If your GetSchedulesAsync method already returns a list, there is no need to use ToListAsync(). You can simply use: return await db.GetSchedulesAsync(start, end);