From 781e604cb7763e7acf4b2dbbff67a10925b6d50e Mon Sep 17 00:00:00 2001 From: shimoncohen Date: Tue, 4 Aug 2026 10:10:28 +0300 Subject: [PATCH 1/3] perf(gpkg): reuse connection, prepared bulk-insert, keyset paging (MAPCO-11320) GpkgClient opened a fresh SQLiteConnection for every tile/batch operation. Hold one lazily-opened connection per client instead, guarded by a lock so the client is safe under the concurrent access Data.GetLastExistingTile (Parallel.ForEachAsync) already performs and that the service-side parallel merge will add. Open it with WAL + synchronous=NORMAL for concurrent reads and far fewer fsyncs on the hot path. - InsertTiles: bind parameters once and reuse a prepared statement per batch instead of re-adding parameters on every row. - GetBatch: keyset pagination on rowid (WHERE rowid > cursor ORDER BY rowid) instead of LIMIT/OFFSET, so page cost stays constant as the merge advances. Signature returns (tiles, lastId); Gpkg persists lastId as the batch identifier for resume. - GpkgClient is now IDisposable and closes the connection; IData : IDisposable and both apps dispose sources+target after the merge to release file locks. Migration note: the gpkg batch identifier changes from an offset to a rowid cursor. Drain in-flight gpkg-source jobs before deploying so resume checkpoints are not reinterpreted across the format change. Co-Authored-By: Claude Opus 4.8 (1M context) --- MergerCli/Program.cs | 7 ++ MergerLogic/Clients/GpkgClient.cs | 133 +++++++++++++------- MergerLogic/Clients/IGpkgClient.cs | 4 +- MergerLogic/DataTypes/Data.cs | 6 + MergerLogic/DataTypes/Gpkg.cs | 44 +++---- MergerLogic/DataTypes/IData.cs | 2 +- MergerLogicUnitTests/DataTypes/GpkgTest.cs | 7 +- MergerLogicUnitTests/Utils/GpkgUtilsTest.cs | 42 ++++++- MergerService/Runners/TaskExecutor.cs | 11 +- 9 files changed, 171 insertions(+), 85 deletions(-) diff --git a/MergerCli/Program.cs b/MergerCli/Program.cs index 32fdc752..6adde8f4 100644 --- a/MergerCli/Program.cs +++ b/MergerCli/Program.cs @@ -136,6 +136,13 @@ private static void Main(string[] args) _logger.LogError(ex, ex.Message); return; } + finally + { + foreach (IData source in sources) + { + source.Dispose(); + } + } totalTimeStopwatch.Stop(); // Get the elapsed time as a TimeSpan value. diff --git a/MergerLogic/Clients/GpkgClient.cs b/MergerLogic/Clients/GpkgClient.cs index 6eb801bc..6b3bd6e9 100644 --- a/MergerLogic/Clients/GpkgClient.cs +++ b/MergerLogic/Clients/GpkgClient.cs @@ -18,6 +18,13 @@ public class GpkgClient : DataUtils, IGpkgClient private readonly ILogger _logger; private readonly IFileSystem _fileSystem; + // Single reusable connection for the hot path (GetTile/TileExists/GetLastTile/InsertTiles/GetBatch). + // Opened lazily and guarded by _connectionLock so the client is safe under the concurrent access + // that Data.GetLastExistingTile (Parallel.ForEachAsync) and the service-side parallel merge perform. + private readonly object _connectionLock = new object(); + private SQLiteConnection? _connection; + private bool _disposed; + public GpkgClient(string path, ITimeUtils timeUtils, ILogger logger, IFileSystem fileSystem, IGeoUtils geoUtils) : base(path, geoUtils) { @@ -27,6 +34,41 @@ public GpkgClient(string path, ITimeUtils timeUtils, ILogger logger, this._tileCache = this.InternalGetTileCache(); } + // Returns the shared connection, opening it (with WAL) on first use. Caller must hold _connectionLock. + private SQLiteConnection GetConnection() + { + if (this._connection == null) + { + var connection = new SQLiteConnection($"Data Source={this.path}"); + connection.Open(); + using (var pragma = connection.CreateCommand()) + { + // WAL + NORMAL sync: concurrent readers alongside a writer and far fewer fsyncs on the hot path. + pragma.CommandText = "PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL;"; + pragma.ExecuteNonQuery(); + } + + this._connection = connection; + } + + return this._connection; + } + + public void Dispose() + { + if (this._disposed) + { + return; + } + + lock (this._connectionLock) + { + this._connection?.Dispose(); + this._connection = null; + this._disposed = true; + } + } + private string InternalGetTileCache() { if (!this.Exist()) @@ -126,12 +168,11 @@ public void UpdateExtent(Extent extent) public override Tile? GetTile(int z, int x, int y) { - byte[]? blob = null; + byte[]? blob; - using (var connection = new SQLiteConnection($"Data Source={this.path}")) + lock (this._connectionLock) { - connection.Open(); - + var connection = this.GetConnection(); using (var command = connection.CreateCommand()) { command.CommandText = @@ -139,7 +180,7 @@ public void UpdateExtent(Extent extent) command.Parameters.AddWithValue("$z", z); command.Parameters.AddWithValue("$x", x); command.Parameters.AddWithValue("$y", y); - blob = (byte[])command.ExecuteScalar(); + blob = command.ExecuteScalar() as byte[]; } } @@ -148,10 +189,9 @@ public void UpdateExtent(Extent extent) public override bool TileExists(int z, int x, int y) { - using (var connection = new SQLiteConnection($"Data Source={this.path}")) + lock (this._connectionLock) { - connection.Open(); - + var connection = this.GetConnection(); using (var command = connection.CreateCommand()) { command.CommandText = @@ -162,73 +202,71 @@ public override bool TileExists(int z, int x, int y) using (var reader = command.ExecuteReader(System.Data.CommandBehavior.SingleRow)) { - // Check if a row was returned - if (reader.HasRows) - { - return true; - } + return reader.HasRows; } } } - - return false; } public void InsertTiles(IEnumerable tiles) { - using (var connection = new SQLiteConnection($"Data Source={this.path}")) + lock (this._connectionLock) { - connection.Open(); - + var connection = this.GetConnection(); + using (var transaction = connection.BeginTransaction()) using (var command = connection.CreateCommand()) { command.CommandText = $"REPLACE INTO \"{this._tileCache}\" (zoom_level, tile_column, tile_row, tile_data) VALUES ($z, $x, $y, $blob)"; - using (var transaction = connection.BeginTransaction()) - { - foreach (Tile tile in tiles) - { - byte[] tileBytes = tile.GetImageBytes(); - SQLiteParameter blobParameter = - new SQLiteParameter("$blob", System.Data.DbType.Binary, tileBytes.Length); - blobParameter.Value = tileBytes; - - command.Parameters.AddWithValue("$z", tile.Z); - command.Parameters.AddWithValue("$x", tile.X); - command.Parameters.AddWithValue("$y", tile.Y); - command.Parameters.Add(blobParameter); - command.ExecuteNonQuery(); - } + // Bind parameters once and reuse the prepared statement for every tile in the batch. + var zParameter = command.Parameters.Add("$z", System.Data.DbType.Int32); + var xParameter = command.Parameters.Add("$x", System.Data.DbType.Int32); + var yParameter = command.Parameters.Add("$y", System.Data.DbType.Int32); + var blobParameter = command.Parameters.Add("$blob", System.Data.DbType.Binary); + command.Prepare(); - transaction.Commit(); + foreach (Tile tile in tiles) + { + zParameter.Value = tile.Z; + xParameter.Value = tile.X; + yParameter.Value = tile.Y; + blobParameter.Value = tile.GetImageBytes(); + command.ExecuteNonQuery(); } + + transaction.Commit(); } } } - public List GetBatch(int batchSize, long offset) + // Keyset pagination over the implicit rowid (present and indexed on every ordinary SQLite table, + // aliasing the INTEGER PRIMARY KEY where one is defined): seeks past the last rowid returned instead + // of scanning and discarding `lastId` rows, so page cost stays constant as the merge progresses. + // `lastId` is the cursor (0 to start); the returned LastId is the greatest rowid read (or the passed + // cursor when the page is empty), to feed the next call. + public (List Tiles, long LastId) GetBatch(int batchSize, long lastId) { - List tiles = new List(); + List tiles = new List(batchSize); - using (var connection = new SQLiteConnection($"Data Source={this.path}")) + lock (this._connectionLock) { - connection.Open(); - + var connection = this.GetConnection(); using (var command = connection.CreateCommand()) { command.CommandText = - $"SELECT zoom_level, tile_column, tile_row, tile_data FROM \"{this._tileCache}\" ORDER BY zoom_level ASC limit $limit offset $offset"; + $"SELECT rowid, zoom_level, tile_column, tile_row, tile_data FROM \"{this._tileCache}\" WHERE rowid > $lastId ORDER BY rowid ASC LIMIT $limit"; + command.Parameters.AddWithValue("$lastId", lastId); command.Parameters.AddWithValue("$limit", batchSize); - command.Parameters.AddWithValue("$offset", offset); using (var reader = command.ExecuteReader()) { while (reader.Read()) { - var z = reader.GetInt32(0); - var x = reader.GetInt32(1); - var y = reader.GetInt32(2); + lastId = reader.GetInt64(0); + var z = reader.GetInt32(1); + var x = reader.GetInt32(2); + var y = reader.GetInt32(3); var blob = (byte[])reader["tile_data"]; Tile tile = this.CreateTile(z, x, y, blob)!; @@ -238,7 +276,7 @@ public List GetBatch(int batchSize, long offset) } } - return tiles; + return (tiles, lastId); } public Tile? GetLastTile(int[] coords, int currentTileZoom) @@ -249,10 +287,9 @@ public List GetBatch(int batchSize, long offset) } Tile? lastTile = null; - using (var connection = new SQLiteConnection($"Data Source={this.path}")) + lock (this._connectionLock) { - connection.Open(); - + var connection = this.GetConnection(); using (var command = connection.CreateCommand()) { // Build command diff --git a/MergerLogic/Clients/IGpkgClient.cs b/MergerLogic/Clients/IGpkgClient.cs index 37b72bc5..a619eec3 100644 --- a/MergerLogic/Clients/IGpkgClient.cs +++ b/MergerLogic/Clients/IGpkgClient.cs @@ -4,9 +4,9 @@ namespace MergerLogic.Clients { - public interface IGpkgClient : IDataUtils + public interface IGpkgClient : IDataUtils, IDisposable { - List GetBatch(int batchSize, long offset); + (List Tiles, long LastId) GetBatch(int batchSize, long lastId); Extent GetExtent(); Tile? GetLastTile(int[] coords, int currentTileZoom); long GetTileCount(); diff --git a/MergerLogic/DataTypes/Data.cs b/MergerLogic/DataTypes/Data.cs index ea6bf8d4..8a9ff9ce 100644 --- a/MergerLogic/DataTypes/Data.cs +++ b/MergerLogic/DataTypes/Data.cs @@ -336,5 +336,11 @@ public virtual void Wrapup() public abstract long TileCount(); public abstract void setBatchIdentifier(string batchIdentifier); + + public virtual void Dispose() + { + (this.Utils as IDisposable)?.Dispose(); + GC.SuppressFinalize(this); + } } } diff --git a/MergerLogic/DataTypes/Gpkg.cs b/MergerLogic/DataTypes/Gpkg.cs index 9b1af8cf..9e73ba22 100644 --- a/MergerLogic/DataTypes/Gpkg.cs +++ b/MergerLogic/DataTypes/Gpkg.cs @@ -8,7 +8,8 @@ namespace MergerLogic.DataTypes { public class Gpkg : Data { - private long _offset; + // Keyset cursor: the greatest tile id handed out so far. Persisted as the batch identifier for resume. + private long _batchCursorId; private Extent _extent; private readonly IConfigurationManager _configManager; static readonly object _locker = new object(); @@ -18,7 +19,7 @@ public Gpkg(IConfigurationManager configuration, IServiceProvider container, : base(container, DataType.GPKG, path, batchSize, grid, origin, isBase, extent) { this._logger.LogDebug($"[{MethodBase.GetCurrentMethod()?.Name}] Ctor started"); - this._offset = 0; + this._batchCursorId = 0; this._configManager = configuration; if (isBase) @@ -82,7 +83,7 @@ public override void Reset() { lock (_locker) { - this._offset = 0; + this._batchCursorId = 0; } } @@ -91,28 +92,19 @@ public override List GetNextBatch(out string currentBatchIdentifier, out s lock (_locker) { this._logger.LogDebug($"[{MethodBase.GetCurrentMethod()?.Name}] started"); - currentBatchIdentifier = this._offset.ToString(); - List tiles = new List(); - if (this._offset != totalTilesCount) - { - //TODO: optimize after IOC refactoring - int counter = 0; - - tiles = this.Utils.GetBatch(this.BatchSize, this._offset) - .Select(t => - { - Tile tile = this.ConvertOriginTile(t); - tile = this.ToCurrentGrid(tile); - counter++; - return tile; - }).Where(t => t != null).ToList(); - - Interlocked.Add(ref this._offset, counter); - nextBatchIdentifier = this._offset.ToString(); - - return tiles; - } - nextBatchIdentifier = this._offset.ToString(); + currentBatchIdentifier = this._batchCursorId.ToString(); + + var (rawTiles, cursor) = this.Utils.GetBatch(this.BatchSize, this._batchCursorId); + List tiles = rawTiles + .Select(t => + { + Tile tile = this.ConvertOriginTile(t); + tile = this.ToCurrentGrid(tile); + return tile; + }).Where(t => t != null).ToList(); + + this._batchCursorId = cursor; + nextBatchIdentifier = this._batchCursorId.ToString(); this._logger.LogDebug($"[{MethodBase.GetCurrentMethod()?.Name}] ended"); return tiles; } @@ -122,7 +114,7 @@ public override void setBatchIdentifier(string batchIdentifier) { lock (_locker) { - this._offset = long.Parse(batchIdentifier); + this._batchCursorId = long.Parse(batchIdentifier); } } diff --git a/MergerLogic/DataTypes/IData.cs b/MergerLogic/DataTypes/IData.cs index c8208823..7d14026f 100644 --- a/MergerLogic/DataTypes/IData.cs +++ b/MergerLogic/DataTypes/IData.cs @@ -2,7 +2,7 @@ namespace MergerLogic.DataTypes { - public interface IData + public interface IData : IDisposable { public DataType Type { get; } public string Path { get; } diff --git a/MergerLogicUnitTests/DataTypes/GpkgTest.cs b/MergerLogicUnitTests/DataTypes/GpkgTest.cs index 814fa62c..3f4c272d 100644 --- a/MergerLogicUnitTests/DataTypes/GpkgTest.cs +++ b/MergerLogicUnitTests/DataTypes/GpkgTest.cs @@ -671,7 +671,7 @@ public void SetBatchIdentifier(bool isOneXOne, bool isBase, GridOrigin origin, i this.SetupRequiredBaseMocks(isBase, isOneXOne, extent); this._gpkgUtilsMock.Setup(utils => utils.GetBatch(10, It.IsAny())) - .Returns(new List()); + .Returns((_, lastId) => (new List(), lastId)); var gpkg = new Gpkg(this._configurationManagerMock.Object, this._serviceProviderMock.Object, "test.gpkg", 10, grid, origin, @@ -709,7 +709,7 @@ public void Reset(bool isOneXOne, bool isBase, GridOrigin origin, int batchSize) this.SetupRequiredBaseMocks(isBase, isOneXOne, extent); this._gpkgUtilsMock.Setup(utils => utils.GetBatch(batchSize, It.IsAny())) - .Returns(new List { new Tile(0, 0, 0, this._jpegImageData) }); + .Returns((_, lastId) => (new List { new Tile(0, 0, 0, this._jpegImageData) }, lastId + 1)); if (origin == GridOrigin.UPPER_LEFT) { this._geoUtilsMock.Setup(converter => converter.FlipY(It.IsAny())) @@ -769,10 +769,11 @@ public void GetNextBatch(bool isOneXOne, bool isBase, GridOrigin origin, int bat var seq = new MockSequence(); for (var i = 0; i < tileBatches.Count; i++) { + var batch = tileBatches[i].ToList(); this._gpkgUtilsMock .InSequence(seq) .Setup(utils => utils.GetBatch(batchSize, It.IsAny())) - .Returns(tileBatches[i].ToList()); + .Returns((_, lastId) => (batch, lastId + batch.Count)); for (var j = 0; j < tileBatches[i].Length; j++) { if (origin == GridOrigin.UPPER_LEFT) diff --git a/MergerLogicUnitTests/Utils/GpkgUtilsTest.cs b/MergerLogicUnitTests/Utils/GpkgUtilsTest.cs index dc5ff40b..4c85314d 100644 --- a/MergerLogicUnitTests/Utils/GpkgUtilsTest.cs +++ b/MergerLogicUnitTests/Utils/GpkgUtilsTest.cs @@ -164,9 +164,13 @@ public void GetBatch(int batchSize, int offset) this._fileSystemMock.Object, this._geoUtilsMock.Object); var comparer = ComparerFactory.Create((t1, t2) => t1?.Z == t2?.Z && t1?.X == t2?.X && t1?.Y == t2?.Y ? 0 : -1); - var res = gpkgUtils.GetBatch(batchSize, offset); - var expected = testTiles.Skip(offset).Take(batchSize); - CollectionAssert.AreEqual(expected.ToArray(), res, comparer); + // Tiles are inserted in order, so their auto-increment ids are contiguous (1..N) and the + // keyset cursor `id > lastId` selects the same rows the old offset-based paging returned. + var (res, cursor) = gpkgUtils.GetBatch(batchSize, offset); + var expected = testTiles.Skip(offset).Take(batchSize).ToArray(); + CollectionAssert.AreEqual(expected, res, comparer); + // Cursor advances to the id of the last row read (or stays put when the page is empty). + Assert.AreEqual(offset + expected.Length, cursor); } this.VerifyAll(); } @@ -191,7 +195,7 @@ public void GetBatchWithLongOffset() var gpkgUtils = new GpkgClient(path, this._timeUtilsMock.Object, this._loggerMock.Object, this._fileSystemMock.Object, this._geoUtilsMock.Object); - var res = gpkgUtils.GetBatch(21, offset); + var (res, _) = gpkgUtils.GetBatch(21, offset); CollectionAssert.AreEqual(Array.Empty(), res); } this.VerifyAll(); @@ -199,6 +203,36 @@ public void GetBatchWithLongOffset() #endregion + #region Dispose + + [TestMethod] + [TestCategory("Dispose")] + public void DisposeAfterUseIsIdempotent() + { + string path = this.GetGpkgPath(); + var testTiles = new Tile[] { new Tile(0, 0, 0, this._jpegImageData) }; + + using (var connection = new SQLiteConnection($"Data Source={path}")) + { + connection.Open(); + this.SetupConstructorRequiredMocks(connection); + this.CreateTestTiles(connection, testTiles); + + var gpkgUtils = new GpkgClient(path, this._timeUtilsMock.Object, this._loggerMock.Object, + this._fileSystemMock.Object, this._geoUtilsMock.Object); + + // Opens the reused connection. + var (res, _) = gpkgUtils.GetBatch(1, 0); + Assert.AreEqual(1, res.Count); + + gpkgUtils.Dispose(); + gpkgUtils.Dispose(); // second dispose must be a no-op, not throw + } + this.VerifyAll(); + } + + #endregion + #region GetExtent public static IEnumerable GenGetExtentParams() diff --git a/MergerService/Runners/TaskExecutor.cs b/MergerService/Runners/TaskExecutor.cs index b63b1a3c..eb1383fb 100644 --- a/MergerService/Runners/TaskExecutor.cs +++ b/MergerService/Runners/TaskExecutor.cs @@ -100,7 +100,8 @@ public void ExecuteTask(MergeTask task, ITaskUtils taskUtils, string? managerCal this._logger.LogDebug($"[{methodName}] BuildDataList"); List sources = this.BuildDataList(metadata.Sources, this._batchMaxSize); - + try + { IData target = sources[0]; target.IsNew = metadata.IsNewTarget; @@ -252,6 +253,14 @@ public void ExecuteTask(MergeTask task, ITaskUtils taskUtils, string? managerCal this._metricsProvider.TilesInBatchGauge(0); } target.Wrapup(); + } + finally + { + foreach (IData source in sources) + { + source.Dispose(); + } + } } this._logger.LogDebug($"[{methodName}] end"); } From 79dc6423a052fb49377e93761f57cba5a1d32e52 Mon Sep 17 00:00:00 2001 From: shimoncohen Date: Wed, 5 Aug 2026 11:05:54 +0300 Subject: [PATCH 2/3] docs: tighten GetBatch keyset-pagination comment Co-Authored-By: Claude Opus 4.8 (1M context) --- MergerLogic/Clients/GpkgClient.cs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/MergerLogic/Clients/GpkgClient.cs b/MergerLogic/Clients/GpkgClient.cs index 6b3bd6e9..baa0a074 100644 --- a/MergerLogic/Clients/GpkgClient.cs +++ b/MergerLogic/Clients/GpkgClient.cs @@ -240,11 +240,9 @@ public void InsertTiles(IEnumerable tiles) } } - // Keyset pagination over the implicit rowid (present and indexed on every ordinary SQLite table, - // aliasing the INTEGER PRIMARY KEY where one is defined): seeks past the last rowid returned instead - // of scanning and discarding `lastId` rows, so page cost stays constant as the merge progresses. - // `lastId` is the cursor (0 to start); the returned LastId is the greatest rowid read (or the passed - // cursor when the page is empty), to feed the next call. + // Keyset pagination over rowid: seeks past lastId instead of OFFSET-scanning, so page cost stays + // constant regardless of depth. lastId is the cursor (0 to start); returns the greatest rowid read + // (or lastId unchanged when the page is empty). public (List Tiles, long LastId) GetBatch(int batchSize, long lastId) { List tiles = new List(batchSize); From 7551a1b6f41bc69502deda7c5153e2f6e54b9ba8 Mon Sep 17 00:00:00 2001 From: shimoncohen Date: Thu, 6 Aug 2026 12:52:46 +0300 Subject: [PATCH 3/3] fix(gpkg): dispose partially-built sources; rename GetConnection Persistent connection reuse turned partial-construction leaks into leaked open SQLite handles. Dispose already-built sources on the CLI < 2 sources early return and when a later CreateDataSource throws in BuildDataList. Rename GetConnection -> GetOrCreateConnection; it opens on first use. Co-Authored-By: Claude Opus 4.8 (1M context) --- MergerCli/Program.cs | 4 ++++ MergerLogic/Clients/GpkgClient.cs | 12 ++++++------ MergerService/Runners/TaskExecutor.cs | 28 +++++++++++++++++++-------- 3 files changed, 30 insertions(+), 14 deletions(-) diff --git a/MergerCli/Program.cs b/MergerCli/Program.cs index 6adde8f4..e98e5c98 100644 --- a/MergerCli/Program.cs +++ b/MergerCli/Program.cs @@ -88,6 +88,10 @@ private static void Main(string[] args) { _logger.LogError("minimum of 2 sources is required"); PrintHelp(programName); + foreach (IData source in sources) + { + source.Dispose(); + } return; } diff --git a/MergerLogic/Clients/GpkgClient.cs b/MergerLogic/Clients/GpkgClient.cs index baa0a074..ed286b4c 100644 --- a/MergerLogic/Clients/GpkgClient.cs +++ b/MergerLogic/Clients/GpkgClient.cs @@ -35,7 +35,7 @@ public GpkgClient(string path, ITimeUtils timeUtils, ILogger logger, } // Returns the shared connection, opening it (with WAL) on first use. Caller must hold _connectionLock. - private SQLiteConnection GetConnection() + private SQLiteConnection GetOrCreateConnection() { if (this._connection == null) { @@ -172,7 +172,7 @@ public void UpdateExtent(Extent extent) lock (this._connectionLock) { - var connection = this.GetConnection(); + var connection = this.GetOrCreateConnection(); using (var command = connection.CreateCommand()) { command.CommandText = @@ -191,7 +191,7 @@ public override bool TileExists(int z, int x, int y) { lock (this._connectionLock) { - var connection = this.GetConnection(); + var connection = this.GetOrCreateConnection(); using (var command = connection.CreateCommand()) { command.CommandText = @@ -212,7 +212,7 @@ public void InsertTiles(IEnumerable tiles) { lock (this._connectionLock) { - var connection = this.GetConnection(); + var connection = this.GetOrCreateConnection(); using (var transaction = connection.BeginTransaction()) using (var command = connection.CreateCommand()) { @@ -249,7 +249,7 @@ public void InsertTiles(IEnumerable tiles) lock (this._connectionLock) { - var connection = this.GetConnection(); + var connection = this.GetOrCreateConnection(); using (var command = connection.CreateCommand()) { command.CommandText = @@ -287,7 +287,7 @@ public void InsertTiles(IEnumerable tiles) Tile? lastTile = null; lock (this._connectionLock) { - var connection = this.GetConnection(); + var connection = this.GetOrCreateConnection(); using (var command = connection.CreateCommand()) { // Build command diff --git a/MergerService/Runners/TaskExecutor.cs b/MergerService/Runners/TaskExecutor.cs index eb1383fb..e5a33972 100644 --- a/MergerService/Runners/TaskExecutor.cs +++ b/MergerService/Runners/TaskExecutor.cs @@ -275,15 +275,27 @@ private List BuildDataList(Source[] paths, int batchSize) if (paths.Length != 0) { - string path = BuildPath(paths[0], true); - sources.Add(this._dataFactory.CreateDataSource(paths[0].Type, path, batchSize, paths[0].Grid, - paths[0].Origin, paths[0].Extent, true)); - foreach (Source source in paths.Skip(1)) + try { - // TODO: add support for HTTP - path = BuildPath(source, false); - sources.Add(this._dataFactory.CreateDataSource(source.Type, path, batchSize, - source.Grid, source.Origin)); + string path = BuildPath(paths[0], true); + sources.Add(this._dataFactory.CreateDataSource(paths[0].Type, path, batchSize, paths[0].Grid, + paths[0].Origin, paths[0].Extent, true)); + foreach (Source source in paths.Skip(1)) + { + // TODO: add support for HTTP + path = BuildPath(source, false); + sources.Add(this._dataFactory.CreateDataSource(source.Type, path, batchSize, + source.Grid, source.Origin)); + } + } + catch + { + // Dispose the sources already built so a mid-list failure doesn't leak open handles. + foreach (IData source in sources) + { + source.Dispose(); + } + throw; } } stopwatch.Stop();