Rewrite the gateway around health, retries, slicing and a Windows service
Scanner - manifest lookup is now depth-bounded and prunes numeric zoom directories, so a scan no longer walks the whole tile payload - rescans only report a version as updated when something actually changed - overlapping runs are refused instead of interleaving their bookkeeping Hot path - token authorisation loads only that token grants and throttles the last_used_at write instead of touching it on every tile - SQLite runs in WAL with a busy timeout and a real connection pool - the catalog lookup used by /tiles no longer loads every basemap Security - the admin surface requires ADMIN_TOKEN (bearer) or ADMIN_USER/ADMIN_PASSWORD (basic), and the service refuses to bind off-host without credentials - CORS is an explicit allow-list; newly minted tokens never reach a redirect URL - client tokens are redacted from the access log Operations - resource health inspection flags a moved or deleted tile root before a 404 - pushes retry with backoff and can be re-queued by hand; the schedule lives in the database so a restart does not lose it - access log plus /metrics counters Image slicing - tile_jobs drive the GDAL bundled with QGIS via scripts/gdal2tiles-qgis.bat - layer_sources remembers what a layer is made of: submit one source to append a flight, or all of them to rebuild the layer as a VRT mosaic with no tile seams; the console exposes both, including a one-click rebuild - slicing is available to integrations through a narrow slice token scope (submit, poll, idempotent by code, completion callback) Console - rebuilt around a fixed rail and single-view navigation with health and outstanding work leading, instead of one long scrolling page Windows service - native install/start/stop/status with crash recovery, auto-start and a rotating log file, so no NSSM wrapper is needed - start-all.bat and stop-all.bat are removed: they killed the process by port with taskkill, which fought the service restart policy Verified with 49 tests and live runs against real tile data and real GDAL.
This commit is contained in:
@@ -7,10 +7,28 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"map-asset-gateway/api-go/internal/uid"
|
||||
"tile-server/api-go/internal/uid"
|
||||
)
|
||||
|
||||
const pushCallbackTimeout = 30 * time.Second
|
||||
|
||||
// Delivery outcomes. retrying is a transient state: the record stays queued
|
||||
// until next_attempt_at elapses.
|
||||
const (
|
||||
pushStatusSucceeded = "succeeded"
|
||||
pushStatusFailed = "failed"
|
||||
pushStatusRetrying = "retrying"
|
||||
|
||||
// pushMaxAttempts caps how many times one record is delivered.
|
||||
pushMaxAttempts = 3
|
||||
// pushRetryBase is the first backoff step; it doubles per attempt.
|
||||
pushRetryBase = 30 * time.Second
|
||||
)
|
||||
|
||||
func (s *Store) CreateTargetSystem(ctx context.Context, input CreateTargetSystemInput) (TargetSystem, error) {
|
||||
@@ -92,6 +110,9 @@ func (s *Store) ListTargetSystems(ctx context.Context) ([]TargetSystem, error) {
|
||||
return items, rows.Err()
|
||||
}
|
||||
|
||||
// PushBasemapVersion queues a downstream notification. The HTTP callback runs
|
||||
// off the request path so a slow target cannot block the caller; the returned
|
||||
// record starts as "pending" and is updated once the callback finishes.
|
||||
func (s *Store) PushBasemapVersion(ctx context.Context, targetCode, basemapCode, version string) (PushRecord, error) {
|
||||
target, err := s.GetTargetSystemByCode(ctx, targetCode)
|
||||
if err != nil {
|
||||
@@ -148,20 +169,6 @@ func (s *Store) PushBasemapVersion(ctx context.Context, targetCode, basemapCode,
|
||||
"pushed_at": toRFC3339(nowUTC()),
|
||||
"targetCode": target.Code,
|
||||
}
|
||||
body, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return PushRecord{}, err
|
||||
}
|
||||
request, err := http.NewRequestWithContext(ctx, target.CallbackMethod, target.CallbackURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return PushRecord{}, err
|
||||
}
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
for key, value := range target.CallbackHeaders {
|
||||
request.Header.Set(key, value)
|
||||
}
|
||||
|
||||
response, err := s.httpClient.Do(request)
|
||||
record := PushRecord{
|
||||
ID: uid.New(),
|
||||
TargetSystemID: target.ID,
|
||||
@@ -169,41 +176,19 @@ func (s *Store) PushBasemapVersion(ctx context.Context, targetCode, basemapCode,
|
||||
BasemapVersionID: versionItem.ID,
|
||||
BasemapCode: basemapItem.Code,
|
||||
BasemapVersion: versionItem.Version,
|
||||
Status: "pending",
|
||||
Request: payload,
|
||||
PushedAt: nowUTC(),
|
||||
}
|
||||
if err != nil {
|
||||
record.Status = "failed"
|
||||
record.ErrorMessage = err.Error()
|
||||
} else {
|
||||
defer response.Body.Close()
|
||||
record.ResponseBody = limitedReadAll(response.Body, 256*1024)
|
||||
status := response.StatusCode
|
||||
record.ResponseStatus = &status
|
||||
if status >= 200 && status < 300 {
|
||||
record.Status = "succeeded"
|
||||
} else {
|
||||
record.Status = "failed"
|
||||
record.ErrorMessage = fmt.Sprintf("target returned status %d", status)
|
||||
}
|
||||
}
|
||||
finishedAt := nowUTC()
|
||||
record.FinishedAt = &finishedAt
|
||||
|
||||
_, insertErr := s.db.ExecContext(ctx, `
|
||||
if _, err := s.db.ExecContext(ctx, `
|
||||
INSERT INTO push_records (
|
||||
id, target_system_id, basemap_version_id, status, request_json, response_status, response_body, error_message, pushed_at, finished_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`, record.ID, record.TargetSystemID, record.BasemapVersionID, record.Status, writeJSON(record.Request), record.ResponseStatus, record.ResponseBody, record.ErrorMessage, toRFC3339(record.PushedAt), nullableTime(record.FinishedAt))
|
||||
if insertErr != nil {
|
||||
return PushRecord{}, insertErr
|
||||
}
|
||||
if err != nil {
|
||||
return record, err
|
||||
}
|
||||
if record.Status != "succeeded" {
|
||||
return record, errors.New(record.ErrorMessage)
|
||||
) VALUES (?, ?, ?, ?, ?, NULL, '', '', ?, NULL)
|
||||
`, record.ID, record.TargetSystemID, record.BasemapVersionID, record.Status, writeJSON(record.Request), toRFC3339(record.PushedAt)); err != nil {
|
||||
return PushRecord{}, err
|
||||
}
|
||||
|
||||
s.enqueuePush("push_records", record.ID, target, payload)
|
||||
return record, nil
|
||||
}
|
||||
|
||||
@@ -223,6 +208,8 @@ func (s *Store) ListVectorPushRecords(ctx context.Context, limit int) ([]VectorP
|
||||
p.response_status,
|
||||
p.response_body,
|
||||
p.error_message,
|
||||
p.attempt_count,
|
||||
p.next_attempt_at,
|
||||
p.pushed_at,
|
||||
p.finished_at
|
||||
FROM vector_push_records p
|
||||
@@ -247,6 +234,8 @@ func (s *Store) ListVectorPushRecords(ctx context.Context, limit int) ([]VectorP
|
||||
return items, rows.Err()
|
||||
}
|
||||
|
||||
// PushVectorAsset queues a downstream notification, mirroring
|
||||
// PushBasemapVersion.
|
||||
func (s *Store) PushVectorAsset(ctx context.Context, targetCode, vectorCode string) (VectorPushRecord, error) {
|
||||
target, err := s.GetTargetSystemByCode(ctx, targetCode)
|
||||
if err != nil {
|
||||
@@ -268,60 +257,240 @@ func (s *Store) PushVectorAsset(ctx context.Context, targetCode, vectorCode stri
|
||||
"pushed_at": toRFC3339(nowUTC()),
|
||||
"targetCode": target.Code,
|
||||
}
|
||||
body, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return VectorPushRecord{}, err
|
||||
}
|
||||
request, err := http.NewRequestWithContext(ctx, target.CallbackMethod, target.CallbackURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return VectorPushRecord{}, err
|
||||
}
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
for key, value := range target.CallbackHeaders {
|
||||
request.Header.Set(key, value)
|
||||
}
|
||||
|
||||
response, err := s.httpClient.Do(request)
|
||||
record := VectorPushRecord{
|
||||
ID: uid.New(),
|
||||
TargetSystemID: target.ID,
|
||||
TargetSystemCode: target.Code,
|
||||
VectorAssetID: asset.ID,
|
||||
VectorCode: asset.Code,
|
||||
Status: "pending",
|
||||
Request: payload,
|
||||
PushedAt: nowUTC(),
|
||||
}
|
||||
if err != nil {
|
||||
record.Status = "failed"
|
||||
record.ErrorMessage = err.Error()
|
||||
} else {
|
||||
defer response.Body.Close()
|
||||
record.ResponseBody = limitedReadAll(response.Body, 256*1024)
|
||||
status := response.StatusCode
|
||||
record.ResponseStatus = &status
|
||||
if status >= 200 && status < 300 {
|
||||
record.Status = "succeeded"
|
||||
} else {
|
||||
record.Status = "failed"
|
||||
record.ErrorMessage = fmt.Sprintf("target returned status %d", status)
|
||||
}
|
||||
}
|
||||
finishedAt := nowUTC()
|
||||
record.FinishedAt = &finishedAt
|
||||
|
||||
_, insertErr := s.db.ExecContext(ctx, `
|
||||
if _, err := s.db.ExecContext(ctx, `
|
||||
INSERT INTO vector_push_records (
|
||||
id, target_system_id, vector_asset_id, status, request_json, response_status, response_body, error_message, pushed_at, finished_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`, record.ID, record.TargetSystemID, record.VectorAssetID, record.Status, writeJSON(record.Request), record.ResponseStatus, record.ResponseBody, record.ErrorMessage, toRFC3339(record.PushedAt), nullableTime(record.FinishedAt))
|
||||
if insertErr != nil {
|
||||
return VectorPushRecord{}, insertErr
|
||||
}
|
||||
if err != nil {
|
||||
return record, err
|
||||
}
|
||||
if record.Status != "succeeded" {
|
||||
return record, errors.New(record.ErrorMessage)
|
||||
) VALUES (?, ?, ?, ?, ?, NULL, '', '', ?, NULL)
|
||||
`, record.ID, record.TargetSystemID, record.VectorAssetID, record.Status, writeJSON(record.Request), toRFC3339(record.PushedAt)); err != nil {
|
||||
return VectorPushRecord{}, err
|
||||
}
|
||||
|
||||
s.enqueuePush("vector_push_records", record.ID, target, payload)
|
||||
return record, nil
|
||||
}
|
||||
|
||||
// enqueuePush performs the first delivery attempt off the request path. A
|
||||
// failure inside the retry budget is rescheduled by dispatchPushRetries.
|
||||
func (s *Store) enqueuePush(table, pushID string, target TargetSystem, payload map[string]any) {
|
||||
s.pushWG.Add(1)
|
||||
go func() {
|
||||
defer s.pushWG.Done()
|
||||
s.attemptPush(table, pushID, 1, target, payload)
|
||||
}()
|
||||
}
|
||||
|
||||
// attemptPush performs one delivery attempt and records the outcome. attempt is
|
||||
// the 1-based attempt number; table is a trusted internal literal.
|
||||
func (s *Store) attemptPush(table, pushID string, attempt int, target TargetSystem, payload map[string]any) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), pushCallbackTimeout)
|
||||
defer cancel()
|
||||
|
||||
status, body, err := s.deliverCallback(ctx, target, payload)
|
||||
errorMessage := ""
|
||||
if err != nil {
|
||||
errorMessage = err.Error()
|
||||
}
|
||||
var responseStatus any
|
||||
if status != 0 {
|
||||
responseStatus = status
|
||||
}
|
||||
|
||||
if err != nil && attempt < pushMaxAttempts {
|
||||
// Stay queued: the dispatcher picks it up once the backoff elapses.
|
||||
next := nowUTC().Add(pushRetryDelay(attempt))
|
||||
if _, updateErr := s.db.ExecContext(context.Background(), fmt.Sprintf(`
|
||||
UPDATE %s
|
||||
SET status = ?, attempt_count = ?, next_attempt_at = ?, response_status = ?, response_body = ?,
|
||||
error_message = ?, finished_at = NULL
|
||||
WHERE id = ?
|
||||
`, table), pushStatusRetrying, attempt, toRFC3339(next), responseStatus, body, errorMessage, pushID); updateErr != nil {
|
||||
log.Printf("reschedule push %s/%s: %v", table, pushID, updateErr)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
recordStatus := pushStatusSucceeded
|
||||
if err != nil {
|
||||
recordStatus = pushStatusFailed
|
||||
}
|
||||
if _, updateErr := s.db.ExecContext(context.Background(), fmt.Sprintf(`
|
||||
UPDATE %s
|
||||
SET status = ?, attempt_count = ?, next_attempt_at = NULL, response_status = ?, response_body = ?,
|
||||
error_message = ?, finished_at = ?
|
||||
WHERE id = ?
|
||||
`, table), recordStatus, attempt, responseStatus, body, errorMessage, toRFC3339(nowUTC()), pushID); updateErr != nil {
|
||||
log.Printf("record push %s/%s: %v", table, pushID, updateErr)
|
||||
}
|
||||
}
|
||||
|
||||
// pushRetryDelay backs off exponentially: 30s, 60s, ...
|
||||
func pushRetryDelay(attempt int) time.Duration {
|
||||
if attempt < 1 {
|
||||
attempt = 1
|
||||
}
|
||||
return pushRetryBase * time.Duration(1<<(attempt-1))
|
||||
}
|
||||
|
||||
// dispatchPushRetries finds attempts whose backoff has elapsed and runs them.
|
||||
func (s *Store) dispatchPushRetries(ctx context.Context) error {
|
||||
for _, table := range []string{"push_records", "vector_push_records"} {
|
||||
due, err := s.dueRetries(ctx, table, 5)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, item := range due {
|
||||
target, err := s.GetTargetSystemByID(ctx, item.targetID)
|
||||
if err != nil {
|
||||
s.failRetry(table, item.id, fmt.Sprintf("target system is gone: %v", err))
|
||||
continue
|
||||
}
|
||||
if !target.Enabled {
|
||||
s.failRetry(table, item.id, "target system is disabled")
|
||||
continue
|
||||
}
|
||||
s.attemptPush(table, item.id, item.attempt+1, target, item.payload)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type duePush struct {
|
||||
id string
|
||||
attempt int
|
||||
targetID string
|
||||
payload map[string]any
|
||||
}
|
||||
|
||||
func (s *Store) dueRetries(ctx context.Context, table string, limit int) ([]duePush, error) {
|
||||
now := toRFC3339(nowUTC())
|
||||
rows, err := s.db.QueryContext(ctx, fmt.Sprintf(`
|
||||
SELECT id, attempt_count, target_system_id, request_json
|
||||
FROM %s
|
||||
WHERE status = ? AND (next_attempt_at IS NULL OR next_attempt_at <= ?)
|
||||
ORDER BY next_attempt_at
|
||||
LIMIT ?
|
||||
`, table), pushStatusRetrying, now, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var items []duePush
|
||||
for rows.Next() {
|
||||
var item duePush
|
||||
var requestJSON string
|
||||
if err := rows.Scan(&item.id, &item.attempt, &item.targetID, &requestJSON); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if strings.TrimSpace(requestJSON) != "" {
|
||||
if err := readMaybeJSON(requestJSON, &item.payload); err != nil {
|
||||
item.payload = nil
|
||||
}
|
||||
}
|
||||
items = append(items, item)
|
||||
}
|
||||
return items, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Store) failRetry(table, pushID, message string) {
|
||||
if _, err := s.db.ExecContext(context.Background(), fmt.Sprintf(`
|
||||
UPDATE %s SET status = ?, error_message = ?, next_attempt_at = NULL, finished_at = ? WHERE id = ?
|
||||
`, table), pushStatusFailed, message, toRFC3339(nowUTC()), pushID); err != nil {
|
||||
log.Printf("fail push %s/%s: %v", table, pushID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// GetTargetSystemByID looks a target up by primary key for retry dispatch.
|
||||
func (s *Store) GetTargetSystemByID(ctx context.Context, id string) (TargetSystem, error) {
|
||||
row := s.db.QueryRowContext(ctx, `
|
||||
SELECT id, code, name, callback_url, callback_method, callback_headers_json, enabled, created_at, updated_at
|
||||
FROM target_systems
|
||||
WHERE id = ?
|
||||
`, strings.TrimSpace(id))
|
||||
item, err := scanTargetSystem(row)
|
||||
if err != nil {
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return TargetSystem{}, fmt.Errorf("target system %q not found", id)
|
||||
}
|
||||
return TargetSystem{}, err
|
||||
}
|
||||
return item, nil
|
||||
}
|
||||
|
||||
// RetryPush re-queues a finished push record. kind is "basemap" or "vector".
|
||||
func (s *Store) RetryPush(ctx context.Context, kind, recordID string) error {
|
||||
table := "push_records"
|
||||
if kind == "vector" {
|
||||
table = "vector_push_records"
|
||||
} else if kind != "basemap" {
|
||||
return fmt.Errorf("unknown push kind %q", kind)
|
||||
}
|
||||
result, err := s.db.ExecContext(ctx, fmt.Sprintf(`
|
||||
UPDATE %s
|
||||
SET status = ?, attempt_count = 0, next_attempt_at = ?, finished_at = NULL, error_message = ''
|
||||
WHERE id = ?
|
||||
`, table), pushStatusRetrying, toRFC3339(nowUTC()), strings.TrimSpace(recordID))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
affected, err := result.RowsAffected()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if affected == 0 {
|
||||
return fmt.Errorf("push record %q not found", recordID)
|
||||
}
|
||||
// Kick the dispatcher so a manual retry does not wait for the next tick.
|
||||
s.pushWG.Add(1)
|
||||
go func() {
|
||||
defer s.pushWG.Done()
|
||||
_ = s.dispatchPushRetries(context.Background())
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// deliverCallback performs the downstream HTTP call for a target system.
|
||||
func (s *Store) deliverCallback(ctx context.Context, target TargetSystem, payload map[string]any) (int, string, error) {
|
||||
return s.postJSON(ctx, target.CallbackMethod, target.CallbackURL, target.CallbackHeaders, payload)
|
||||
}
|
||||
|
||||
// postJSON sends a JSON payload and returns the status code plus a truncated
|
||||
// response body. Shared by push callbacks and tile-job completion callbacks.
|
||||
func (s *Store) postJSON(ctx context.Context, method, url string, headers map[string]string, payload map[string]any) (int, string, error) {
|
||||
body, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return 0, "", err
|
||||
}
|
||||
request, err := http.NewRequestWithContext(ctx, method, url, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return 0, "", err
|
||||
}
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
for key, value := range headers {
|
||||
request.Header.Set(key, value)
|
||||
}
|
||||
|
||||
response, err := s.httpClient.Do(request)
|
||||
if err != nil {
|
||||
return 0, "", err
|
||||
}
|
||||
defer response.Body.Close()
|
||||
|
||||
responseBody := limitedReadAll(response.Body, 256*1024)
|
||||
// Drain whatever is left so the connection can be reused.
|
||||
_, _ = io.Copy(io.Discard, response.Body)
|
||||
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
||||
return response.StatusCode, responseBody, fmt.Errorf("target returned status %d", response.StatusCode)
|
||||
}
|
||||
return response.StatusCode, responseBody, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user