Modes d'ingestion

Le champ ingest.mode d'une source de données indique ce qui pilote la récupération. Tout le reste — le bloc request, response.itemsPath, mapping, filter, table.columns — fonctionne de la même manière dans tous les modes : passer de l'un à l'autre est un changement de pilote, pas une réécriture.

table.timestampColumn est requis pour les modes qui écrivent des lignes, car le stockage partitionne dessus. on-demand peut utiliser la colonne système __fetchedAt. Les sources query obtiennent leur schéma depuis les métadonnées du pilote distant. table.columns est requis dans les autres modes déclaratifs — c'est le seul validateur de lignes dont dispose une source déclarative.

Installer et configurer une source

L'installation d'une extension, la configuration de sa connexion et l'exécution de la source sont des étapes distinctes. Les administrateurs de l'organisation installent l'extension et gèrent ses connexions. Les administrateurs autorisés de l'espace de travail activent les sources dans l'environnement concerné. L'installation seule ne vérifie pas la connexion distante ni l'ingestion.

Dans la vue Connections de la source installée, sélectionnez Configure pour ouvrir l'onglet Connections de l'extension. Créez ou sélectionnez la connexion applicable à l'environnement voulu. Les identifiants appartiennent à l'organisation ; les connexions peuvent cibler Development, Production ou, explicitement, tous les environnements. Utilisez des connexions séparées lorsque les identifiants diffèrent. La promotion d'une configuration vers Production ne copie pas les identifiants de Development. Les règles existantes de sélection des connexions et les autorisations administrateur continuent de s'appliquer. Les sources de base de données query exigent une connexion basic-auth active ciblant l'environnement exact ; une connexion commune à tous les environnements ne satisfait pas cette exigence.

L'onglet Setup décrit l'installation au niveau de l'organisation. L'onglet Connections indique si les métadonnées des identifiants applicables sont configurées ; cela ne vérifie pas la connectivité distante. Consultez séparément la dernière opération et la fraîcheur des données dans Data Sources, puis Open table pour voir les enregistrements persistés. Modifiez les connexions et renouvelez les identifiants via les mêmes contrôles que pour les autres extensions ; n'intégrez pas de secrets dans le manifeste d'une source.

Si le formulaire ne se charge pas, utilisez l'action de nouvelle tentative ou demandez à un administrateur de vérifier le fournisseur d'identifiants déclaré par l'extension. Un échec de chargement ne signifie pas qu'aucun identifiant n'est nécessaire.

manual — importer dans une source

Une source manuelle déclare le format, la sélection, le mapping et la validation du fichier importé. Elle possède sa table gérée, contrairement à une table utilisateur générique. Déclarez mode: 'manual', upload: { itemsPath: '$', format: 'csv' }, asOfColumn: 'asOf', le mapping et le writeMode. La table doit déclarer une colonne asOf de type timestamp et un timestampColumn de partitionnement. asOfColumn ne peut pas aussi apparaître dans mapping : le fichier ne peut pas modifier cet horodatage.

Dans Analytics, ouvrez le détail de la source installée, sélectionnez Fichier source, saisissez Date de référence (UTC) et choisissez Envoyer vers la source. Le mapping de la source est appliqué au fichier stocké ; les lignes valides sont écrites dans cette source et cet environnement précis, et le nombre de lignes rejetées est affiché. Limites : 16 MiB et 50 000 lignes sélectionnées ; le formulaire refuse un fichier plus volumineux avant de l'importer. Le fichier parvient à l'ingestion dans une seule requête, bornée aux mêmes 16 MiB : un fichier juste sous la limite dont les guillemets et les sauts de ligne doivent être échappés (surtout en JSON) peut encore être refusé comme trop volumineux. Un fichier qui dépasse l'une ou l'autre limite est refusé en entier, jamais réduit à ses premières lignes : découpez-le en plusieurs imports, ou faites-le envoyer par son producteur en livraisons push de 1 000 lignes au plus. Une marque d'ordre des octets UTF-8 en début de fichier, que le format CSV UTF-8 d'Excel écrit, est ignorée. La provenance conserve la source, la clé de stockage, la version du descripteur, l'heure de réception et l'heure as-of. Réessayer avec le même fichier stocké et la même heure as-of réutilise le reçu de la source ; changer le contenu ou l'heure as-of sous ce reçu est refusé. Renvoyer le même fichier indique Ce fichier a déjà été reçu ; rien de nouveau à écrire. : ses lignes ont été écrites à sa première réception, et rien n'est écrit deux fois. La Dernière synchronisation de la source et son nombre de lignes restent ceux de ce premier envoi.

Un import replace contenant des lignes invalides échoue avant de remplacer les données existantes.

Une source manuelle est aussi, pour l'instant, la voie d'entrée de deux fichiers qu'une source poll ne peut pas récupérer :

• Un fichier publié sous une URL qui change chaque jour, comme celui de GPSJam, https://gpsjam.org/data/2026-09-27-h3_4.csv. L'URL d'une requête poll est fixée dans le manifeste et n'a pas d'emplacement pour une date : téléchargez le fichier du jour et importez-le avec ce jour comme Date de référence (UTC).
• Un fichier de plus de 10 000 lignes. Une réponse poll ou on-demand qui en produit davantage est refusée en entier (voir Limites de ressources) ; un import manuel en accepte jusqu'à 50 000.

on-demand — récupérer, ajouter et renvoyer

Une source on-demand répond à une question que l'utilisateur vient de poser : « quels hôtes correspondent à cette requête ? » Elle n'a pas de planification. Chaque exécution ajoute les lignes validées à la table de la source et renvoie à l'appelant les lignes exactement telles qu'elles ont été stockées : une ligne écartée par votre politique DLP n'est pas renvoyée, pas plus qu'une ligne qui n'a pas pu être enregistrée. Les colonnes visibles __fetchedAt (timestamp), __fetchedBy (utilisateur ou service vérifié) et __args (arguments validés en JSON canonique) sont réservées, sans distinction de casse : ne les déclarez pas et ne les mappez pas. Leur classification est au minimum interne et hérite des niveaux plus stricts de la source. Les tables compatibles existantes reçoivent les colonnes manquantes ; les anciennes lignes conservent des valeurs nulles. Les instantanés installés hérités sans horodatage utilisent __fetchedAt sans que l'instantané soit réécrit.

Une réponse qui produit plus de 10 000 lignes fait échouer la récupération avec exceeds maxRows (10000) : rien n'est renvoyé ni stocké, plutôt que les 10 000 premières. Restreignez la requête par un paramètre, par exemple une taille de page plus petite.

Le résultat persistence indique committed, failed ou skipped; rowsWritten compte uniquement les nouvelles écritures confirmées. Un rejeu du reçu compte zéro. Un échec conserve les lignes récupérées et le nombre des lots déjà confirmés; un délai dépassé ne prouve pas que rien n'a été stocké. Les lots sont limités à 1 000 lignes et 1 Mio, enveloppe comprise. Une ligne trop volumineuse fait échouer l'enregistrement visiblement. Un runtime de récupération seule, sans destination composée, indique skipped.

defineDataSource({
  kind: 'table',
  id: 'shodan-host-search',
  vendor: 'shodan',
  displayName: 'Shodan — host search',
  scope: 'org',
  credentialRef: 'apiKey',
  table: {
    name: 'shodan_host',
    primaryKey: ['ip'],
    timestampColumn: '__fetchedAt',
    columns: [
      { name: 'ip', dataType: 'string', isPrimaryKey: true },
      { name: 'port', dataType: 'int', nullable: true },
      { name: 'org', dataType: 'string', nullable: true },
    ],
  },
  ingest: {
    mode: 'on-demand',
    request: {
      url: 'https://api.shodan.io/shodan/host/search',
      authRef: 'apiKey',
      auth: { placement: 'query', name: 'key' },
    },
    response: { itemsPath: '$.matches' },
    params: {
      query: { in: 'query', name: 'query', required: true, maxLength: 512 },
      page: { in: 'query', type: 'number' },
    },
    mapping: {
      ip: { path: 'ip_str' },
      port: { path: 'port' },
      org: { path: 'org', transform: 'trim_to_null' },
    },
  },
})

Écrire intervalSec ou writeMode sur une source on-demand est une erreur de validation, et non un champ silencieusement ignoré : il n'y a pas de planification et les écritures sont toujours en ajout.

params — ce qu'un appelant peut envoyer

Dans Analytics, ouvrez le panneau de détail de la source et choisissez Fetch now. Saisissez un objet JSON dans Fetch arguments (JSON) avec les clés déclarées dans params, par exemple {"query":"port:443","page":1}. Le panneau liste chaque argument déclaré, s'il est obligatoire et, pour un enum, ses valeurs acceptées ; un argument obligatoire dont l'enum n'a qu'une valeur est prérempli. Une source sans arguments utilise {}. Un argument refusé est nommé dans l'erreur, avec ses valeurs acceptées. Une récupération en échec en donne la raison dans les mots qu'une exécution planifiée enregistre pour le même échec, par exemple que le nom d'hôte de la requête est introuvable, ou le statut HTTP et le message d'erreur du fournisseur. Le panneau ne demande de réessayer que lorsque la cause n'a pas pu être nommée. Le panneau affiche les lignes retournées sous forme de grille, sous les colonnes que la source déclare dans table.columns, et le nombre de lignes rejetées, ainsi qu'un résultat vide explicite. Chaque clic demande des données fraîches. L'aperçu disparaît à la fermeture du panneau ou au changement de source ; les lignes stockées restent soumises à la rétention de la table. Le panneau présente les écritures confirmées, les échecs partiels, l'historique et le lien Open table après création. Les réponses portent Cache-Control: no-store. Une récupération qui échoue est enregistrée sur la source comme l'échec d'une exécution planifiée : son statut, sa dernière erreur et son Activité indiquent pourquoi.

params est la liste fermée des arguments qu'un appelant peut fournir. Un argument que vous n'avez pas déclaré est rejeté, jamais transmis. Chaque entrée indique où va la valeur et ce qu'elle peut être :

Un argument d'appelant ne peut jamais devenir autre chose qu'un paramètre de requête ou un champ du corps JSON. Il ne peut jamais atteindre le schéma, l'hôte ou le chemin de l'URL de la requête : il ne peut donc pas rediriger l'appel vers un autre serveur. Deux règles sont appliquées lors de l'analyse du manifeste plutôt qu'au moment de l'appel :

• un paramètre dont le nom sur le fil est celui qu'utilise request.auth pour un placement query est rejeté — sinon un appelant pourrait choisir la valeur du credential ;
• un paramètre dont le nom sur le fil entre en collision avec une clé statique de request.query est rejeté — sinon il l'écraserait silencieusement.

in: 'body' exige request.method: 'POST' avec un request.body de kind: 'json' dont la value est un objet ; l'argument y est fusionné comme champ de premier niveau.

Il n'existe pas de in: 'path' : une API qui attend son entrée dans le chemin de l'URL ne peut donc pas la recevoir d'un appelant aujourd'hui. Le géocodage en texte libre de TomTom, GET https://api.tomtom.com/search/2/geocode/{query}.json, en est un exemple : {query} dans request.url est envoyé tel quel. Utilisez la variante à paramètres de requête du fournisseur lorsqu'elle existe — pour TomTom, le géocodage structuré GET https://api.tomtom.com/search/2/structuredGeocode.json, dont countryCode, streetNumber, streetName, municipality et postalCode sont tous des paramètres de requête qu'une entrée params peut déclarer.

Ce que l'appelant reçoit en retour

Des lignes, validées contre table.columns, ainsi que le nombre de lignes produites par la source qui ont échoué à cette validation. Les lignes en échec sont écartées, pas renvoyées.

Si la source référence un credential stocké et que l'organisation n'a aucune connexion activée pour celui-ci, la récupération est refusée avec data_source_connection_required — le même refus que celui qu'une exécution planifiée d'une source poll enregistre comme état de synchronisation — plutôt que de s'exécuter sans authentification.

stream — un abonnement maintenu ouvert

ingest: {
  mode: 'stream',
  transport: 'websocket',
  endpoint: {
    url: 'wss://stream.example.com/v1/positions',
    authRef: 'apiKey',
    auth: { placement: 'bearer' },
  },
  subscribe: [{ kind: 'json', value: { BoundingBoxes: [[[-90, -180], [90, 180]]] } }],
  message: { itemsPath: '$' },
  filter: { requireNonNull: ['MetaData.time_utc'] },
  mapping: {
    mmsi: { path: 'MetaData.MMSI' },
    seenAt: { path: 'MetaData.time_utc', transform: 'iso_from_go_time' },
  },
}

• transport vaut websocket — le seul transport doté d'un hôte gouverné.
• endpoint.url doit être en wss://. Le ws:// en clair est refusé pour la même raison que request.url doit être en https://. Ou bien lisez l'hôte depuis la connexion : endpoint: { endpointRef: 'baseUrl', path: '/api/websocket' } ouvre wss:// sur l'endpoint https:// de la connexion — voir Endpoints fournis par une connexion.
• endpoint réutilise le vocabulaire de credentials du bloc request en mode poll : authRef plus un placement auth valant bearer, header ou query — et, pour un stream seulement, frame (ci-dessous). Un en-tête Authorization en clair y est également refusé.
• endpoint.protocols propose des sous-protocoles WebSocket lors de la poignée de main (8 au maximum). Chacun doit être un jeton unique (sans espace ni virgule) et n'apparaître qu'une fois. Si vous en proposez, le serveur doit en choisir un, sinon la connexion est refusée.
• La socket ne se connecte qu'à l'adresse approuvée par le contrôle de sortie et vérifie le certificat du serveur par rapport au nom d'hôte du point de terminaison. Les redirections ne sont pas suivies et la compression n'est pas négociée. Un message de plus de 1 Mio ferme la connexion.
• subscribe est la liste des trames envoyées dès l'ouverture de la socket — { kind: 'json', value } ou { kind: 'text', value }. Une trame WebSocket ne porte aucun type de contenu : contrairement à request.body, la trame text n'a donc pas besoin de contentType.
• message est appliqué à chaque message entrant exactement comme response l'est au corps d'une réponse poll : décoder selon format, puis sélectionner le tableau de lignes à itemsPath. Un message qui est un objet unique plutôt qu'une liste utilise itemsPath: '$', et filter.requireNonNull écarte une trame de contrôle sans données, comme une confirmation d'abonnement.

La politique de reconnexion et de backoff appartient au runtime ; le manifeste n'a aucun paramètre de réglage.

Un stream suit sa connexion. Quand un administrateur enregistre une nouvelle clé sur la connexion qu'il utilise, rattache la source à une autre connexion, change le point de terminaison de la connexion ou supprime la connexion, le stream en cours s'en aperçoit en 10 secondes environ et se reconnecte avec ce que la connexion contient désormais. Les lignes déjà reçues sont validées d'abord : rien n'est perdu ni écrit deux fois. Renommer une connexion, ou le renouvellement d'un jeton OAuth, ne le reconnecte pas. Il n'est jamais nécessaire de mettre une source en pause puis de la reprendre pour qu'un changement de connexion s'applique.

Une ligne reçue est conservée de façon durable en 250 ms au plus, puis validée dans la table avec les lignes qui l'entourent : la table d'un stream reçoit au plus une validation par minute, donc une ligne y apparaît jusqu'à environ une minute après son arrivée. Une validation contient jusqu'à 1 000 lignes et un peu moins de 1 Mio de données de lignes ; un flux assez chargé pour en remplir une plus tôt la valide aussitôt. Un message qui produit plus de 1 000 lignes, ou plus de données de lignes qu'une validation n'en contient, est écarté en entier.

Un message ou une ligne écartés sont conservés tels qu'ils sont arrivés, pendant une heure : votre opérateur Scrydon peut lire ce qui a été écarté pendant cette heure. Ensuite, ce qui a été écarté est supprimé ; son nombre reste dans les statistiques de la source. Un message de plus de 1 Mio ne peut pas être conservé en entier : son nombre de lignes, son motif et son empreinte sont gardés, et l'enregistrement indique que son contenu a été omis.

Un stream est livré à sa table au moins une fois : après une reconnexion, ou quand un autre réplica reprend le stream, un lot peut être validé une seconde fois. Gardez le writeMode: 'upsert' par défaut (ou changed-only) avec un table.primaryKey qui identifie une observation, afin qu'une ligne rejouée s'écrase elle-même. append stockerait le rejeu deux fois : un stream déclaré append ne s'exécute donc que sur un déploiement dont l'opérateur a activé ingestBuffer.recoverableSinkEnabled ; ailleurs, il ne démarre pas, et la source apparaît indisponible, en nommant ce réglage.

Une clé qui voyage dans le premier message

Certains flux (aisstream.io, par exemple) attendent leur clé d'API dans la trame d'abonnement, pas dans un en-tête. Les trames JSON peuvent référencer apiKey ou accessToken stockés, et auth: { placement: 'frame' } garde la clé uniquement dans ces trames — la poignée de main WebSocket ne la transporte nulle part, ni en en-tête ni dans l'URL :

endpoint: {
  url: 'wss://stream.aisstream.io/v0/stream',
  authRef: 'aisstreamKey',          // le credential de l'extension (auth.credentials)
  auth: { placement: 'frame' },     // la clé va dans la trame ci-dessous, nulle part ailleurs
},
subscribe: [
  {
    kind: 'json',
    value: { APIKey: null, BoundingBoxes: [[[51.0, 1.5], [56.0, 9.0]]] },
    credentials: [{ path: ['APIKey'], credential: 'apiKey' }],   // APIKey reçoit son apiKey
  },
],

• authRef nomme le credential ; credential, dans chaque référence, nomme le champ de ce credential qui remplit le chemin (apiKey, ou accessToken pour une connexion OAuth).
• Chaque chemin doit viser un champ d'objet existant et null. Les chemins absents, qui se chevauchent, traversent un tableau ou visent un prototype sont refusés. Les trames texte ne substituent jamais de credentials et aucune valeur secrète n'est livrée dans le manifeste.
• Les références exigent endpoint.authRef, qui nomme le credential.
• Une référence de trame sans endpoint.auth est refusée à la construction et au téléversement de l'extension : le placement par défaut enverrait aussi la même clé en Authorization: Bearer lors de la poignée de main. Indiquez 'frame', ou nommez le placement de poignée de main que vous voulez aussi. Un stream installé avant cette règle continue de fonctionner exactement comme à son installation — la clé dans la trame et sur la poignée de main — jusqu'à ce que vous installiez une version qui indique 'frame'.
• placement: 'frame' est refusé sur un stream dont aucune trame ne référence de credential, ainsi que sur toute requête poll ou on-demand — une requête HTTP n'a pas de trame.

Quand un stream va mal

Un stream enregistre son état de synchronisation une fois par fenêtre de 30 secondes plutôt qu'à chaque message, et indique le problème dans le statut et la dernière erreur de la source :

Une trame écartée par votre filter (confirmation d'abonnement, heartbeat) n'est ni l'un ni l'autre : elle compte comme un message reçu, rien de plus. Aucune valeur de ligne n'apparaît dans ces textes.

Toute source s'exécute sur le service d'ingestion de Scrydon, que chaque déploiement exécute ; les sources stream et push exigent en plus son tampon d'ingestion, pas les sources poll. Un déploiement sans ce tampon affiche chaque source stream et push comme indisponible, avec la marche à suivre pour votre opérateur Scrydon (il s'active avec ingestBuffer.enabled dans les valeurs Helm).

Exemple complet : positions AIS d'aisstream.io

Cette source conserve les positions de navires en direct pour la mer du Nord, depuis aisstream.io, qui diffuse l'AIS par WebSocket et attend sa clé d'API dans la trame d'abonnement. L'extension déclare la clé une seule fois :

import { defineExtension } from '@scrydon/sdk-authoring/extensions'
import { aisPositions } from './data-source-ais-positions'

export default defineExtension({
  id: 'aisstream',
  name: 'AIS Stream',
  version: '1.0.0',
  logo: './assets/icon.svg',
  auth: {
    credentials: { aisstreamKey: { type: 'apiKey', label: 'aisstream.io API key' } },
    default: 'aisstreamKey',
  },
  provides: { dataSources: [aisPositions] },
})

import { defineDataSource } from '@scrydon/sdk-authoring/extensions'

export const aisPositions = defineDataSource({
  kind: 'table',
  id: 'ais-positions-north-sea',
  vendor: 'aisstream',                  // l'id de l'extension : ses connexions portent la clé
  displayName: 'AIS positions — North Sea',
  scope: 'org',
  table: {
    name: 'ais_position',
    primaryKey: ['mmsi', 'observedAt'],  // une ligne par navire et par rapport
    timestampColumn: 'observedAt',
    ttlSec: 7 * 86400,
    columns: [
      { name: 'mmsi', dataType: 'bigint', isPrimaryKey: true },
      { name: 'shipName', dataType: 'string', nullable: true },
      { name: 'latitude', dataType: 'double' },
      { name: 'longitude', dataType: 'double' },
      { name: 'speedKnots', dataType: 'double', nullable: true },
      { name: 'courseDeg', dataType: 'double', nullable: true },
      { name: 'headingDeg', dataType: 'int', nullable: true },
      { name: 'navStatus', dataType: 'int', nullable: true },
      { name: 'observedAt', dataType: 'timestamp', isPrimaryKey: true },
    ],
  },
  ingest: {
    mode: 'stream',
    transport: 'websocket',
    endpoint: {
      url: 'wss://stream.aisstream.io/v0/stream',
      authRef: 'aisstreamKey',
      auth: { placement: 'frame' },
    },
    subscribe: [
      {
        kind: 'json',
        value: {
          APIKey: null,
          BoundingBoxes: [[[51.0, 1.5], [56.0, 9.0]]],   // [[lat, lon], [lat, lon]]
          FilterMessageTypes: ['PositionReport'],
        },
        credentials: [{ path: ['APIKey'], credential: 'apiKey' }],
      },
    ],
    message: { itemsPath: '$', format: 'json' },
    filter: { requireNonNull: ['MetaData.MMSI', 'MetaData.time_utc'] },
    mapping: {
      mmsi: { path: 'MetaData.MMSI' },
      shipName: { path: 'MetaData.ShipName', transform: 'trim_to_null' },
      latitude: { path: 'MetaData.latitude' },
      longitude: { path: 'MetaData.longitude' },
      speedKnots: {
        path: 'Message.PositionReport.Sog',
        transform: 'value_map',
        args: { map: { '102.3': null }, passthrough: 'number' },
      },
      courseDeg: {
        path: 'Message.PositionReport.Cog',
        transform: 'value_map',
        args: { map: { '360': null }, passthrough: 'number' },
      },
      headingDeg: {
        path: 'Message.PositionReport.TrueHeading',
        transform: 'value_map',
        args: { map: { '511': null }, passthrough: 'number' },
      },
      navStatus: { path: 'Message.PositionReport.NavigationalStatus' },
      observedAt: { path: 'MetaData.time_utc', transform: 'iso_from_go_time' },
    },
    writeMode: 'upsert',
  },
})

Chaque message qu'aisstream envoie pour cet abonnement est une seule enveloppe, pas une liste :

{
  "MetaData": {
    "MMSI": 246055000, "MMSI_String": 246055000, "ShipName": "TS10 HILLEGONDA     ",
    "latitude": 53.2949, "longitude": 5.4693,
    "time_utc": "2026-09-28 08:06:58.965887092 +0000 UTC"
  },
  "MessageType": "PositionReport",
  "Message": {
    "PositionReport": {
      "UserID": 246055000, "Latitude": 53.29490166666667, "Longitude": 5.469303333333333,
      "Sog": 3.2, "Cog": 70.2, "TrueHeading": 57, "NavigationalStatus": 0
    }
  }
}

(Le rapport de position contient plus de champs que ceux montrés ici.) Pourquoi chaque partie de la source est écrite ainsi :

• La clé n'atteint jamais la poignée de main. placement: 'frame' la place dans APIKey dans la trame d'abonnement et nulle part ailleurs. Le manifeste porte null ; la connexion de l'organisation fournit la clé quand le stream se connecte.
• itemsPath: '$' sélectionne l'enveloppe elle-même comme une liste d'une ligne : chaque message donne une ligne. aisstream envoie son JSON dans des trames WebSocket binaires ; elles sont lues en UTF-8, et format: 'json' indique comment les décoder, puisqu'un message WebSocket n'a pas de type de contenu.
• Le filter écarte la confirmation d'abonnement. Le premier message de chaque abonnement est {"Message":{"CompressionEnabled":true},"MessageType":"SubscriptionConfirmation"}. Il n'a pas de MetaData et il est donc écarté. Il compte comme un message reçu, jamais comme une ligne écartée par la validation.
• La casse est celle d'aisstream, pas une convention. MetaData écrit latitude et longitude en minuscules, à côté de MMSI, ShipName et time_utc. Un chemin est sensible à la casse : MetaData.Latitude ne résout rien et chaque ligne est écartée faute de latitude. MMSI est un nombre JSON, d'où bigint. Les coordonnées de MetaData sont la position du rapport arrondie à 5 décimales (environ 1 m) ; mappez Message.PositionReport.Latitude et Longitude si vous avez besoin de tous les chiffres. ShipName arrive tantôt complété d'espaces jusqu'à 20 caractères, tantôt rogné ; trim_to_null lui donne une seule écriture et transforme un nom vide en null.
• time_utc n'est pas du RFC 3339. C'est le texte time.Time de Go ; mappé tel quel, chaque ligne est écartée avec observedAt: Invalid ISO datetime. iso_from_go_time le convertit en 2026-09-28T08:06:58.965887092Z et conserve tous les chiffres de la fraction.
• L'AIS envoie « non disponible » sous forme de nombre. Un cap vrai de 511, une route de 360 et une vitesse de 102.3 signifient que le transpondeur n'en a pas transmis. value_map avec passthrough: 'number' transforme exactement ces valeurs en null et laisse passer tous les autres nombres. Elles sont fréquentes : dans un échantillon de rapports de position en mer du Nord, 66 % avaient un cap 511 et 30 % une route 360. Sans ce mappage, elles seraient enregistrées comme de vrais relèvements.
• primaryKey + upsert. La paire mmsi + observedAt conserve la trace de chaque navire, une ligne par rapport, et fait qu'un lot validé deux fois s'écrase lui-même.

Cette source a été vérifiée sur 4 491 rapports de position capturés en direct depuis cet abonnement le 28 septembre 2026 : chaque rapport a produit une ligne valide, et le seul message écarté était la confirmation d'abonnement. Ce matin-là, la même zone a transmis environ 37 rapports de position par seconde, avec un pic à 73, ce que le traitement par lots absorbe. Une zone plus grande ou davantage de types de messages multiplient ce débit ; restreignez BoundingBoxes et FilterMessageTypes à ce que vous utilisez.

Ce que vous voyez dans Analytics → Data sources pour cette source :

push — le fournisseur nous livre les données

Une source push n'a ni request ni endpoint — nous n'émettons aucun appel sortant. Ce qu'elle déclare, c'est la manière dont une livraison est authentifiée, avec le même vocabulaire que le vérificateur de webhooks entrants de Scrydon :

ingest: {
  mode: 'push',
  deliveryIdHeader: 'x-delivery-id',
  auth: {
    mode: 'hmac',
    headerName: 'x-hub-signature-256',
    algorithm: 'sha256',
    timestampHeader: 'x-signature-timestamp',
  },
  delivery: { itemsPath: '$.events' },
  mapping: { id: { path: 'id' }, seenAt: { path: 'timestamp' } },
}

Le manifeste déclare la forme du credential, jamais sa valeur. Il n'existe aucun champ où placer un secret : le secret est le credential de connexion stocké par l'organisation, résolu côté serveur. Préférez hmac, puis header ; ne recourez à url_token que lorsque le producteur ne peut pas envoyer d'en-têtes du tout, car un secret dans une URL transite par les journaux des proxys et les en-têtes Referer.

Une signature hmac couvre davantage que le corps. Le producteur signe la chaîne <horodatage>.<identifiant de livraison>.<corps brut>, où <horodatage> est l'heure Unix en secondes qu'il envoie dans timestampHeader et <identifiant de livraison> la valeur de deliveryIdHeader. Chaque signature est ainsi liée à une livraison, à un instant donné :

• Une livraison dont l'horodatage s'écarte de plus de 300 secondes de l'horloge du récepteur, dans un sens ou dans l'autre, est refusée avec 401 push_signature_stale.
• Un horodatage absent ou non numérique, ou une signature calculée sur le seul corps, est refusé avec 401 push_authentication_failed.
• Une livraison capturée puis renvoyée sous un nouvel identifiant ne se vérifie plus, et un renvoi sous le même identifiant est reconnu comme doublon du reçu d'origine.

timestampHeader doit désigner son propre en-tête : ni l'en-tête de signature, ni celui de l'identifiant de livraison.

Seule une livraison authentifiée est décomptée de la limite de débit push de la source : un appelant qui connaît l'URL de réception mais pas le secret ne peut pas l'épuiser.

Une livraison cesse de compter dans la capacité du tampon de la source dès qu'elle est traitée, mais son identifiant est conservé une heure : pendant ce délai, un renvoi est traité comme doublon du reçu d'origine ; ensuite, l'identifiant est oublié. Cette heure dépasse le double de la fenêtre de 600 secondes pendant laquelle un même horodatage signé est accepté : une livraison signée ne peut donc pas être rejouée une fois son identifiant oublié.

delivery est appliqué à chaque corps livré exactement comme response l'est au corps d'une réponse poll.

Où un émetteur livre

Chaque instance installée d'une source push a sa propre adresse, POST <hôte push public>/api/platform/data-sources/push/<ID de liaison>. L'ID de liaison est dérivé de l'organisation, de l'instance et de son environnement : une instance ajoutée a donc le sien. Un administrateur de l'organisation le trouve dans l'onglet Connexions de l'extension : sur la ligne de la source, ⋯ → Point de terminaison push affiche le chemin, l'en-tête qui porte l'identifiant ou la signature, et l'en-tête d'ID de livraison, chacun avec un bouton de copie. L'hôte push public est celui du déploiement — voir apiIngestion pour la façon dont chaque mode de routage le publie. Platform n'est pas configurée avec cet hôte : la boîte de dialogue n'affiche que le chemin.

Exemple complet : relevés de capteurs signés

Un émetteur envoie des lots de relevés de capteurs et signe chaque corps avec un secret partagé avec l'organisation. L'extension déclare ce secret, et la source le nomme avec credentialRef, pour que le formulaire de connexion le demande :

import { defineExtension } from '@scrydon/sdk-authoring/extensions'
import { sensorReadings } from './data-source-sensor-readings'

export default defineExtension({
  id: 'acme-sensors',
  name: 'Acme Sensors',
  version: '1.0.0',
  logo: './assets/icon.svg',
  auth: {
    credentials: { signingSecret: { type: 'apiKey', label: 'Webhook signing secret' } },
    default: 'signingSecret',
  },
  provides: { dataSources: [sensorReadings] },
})

import { defineDataSource } from '@scrydon/sdk-authoring/extensions'

export const sensorReadings = defineDataSource({
  kind: 'table',
  id: 'acme-sensor-readings',
  vendor: 'acme-sensors',
  displayName: 'Acme — sensor readings',
  scope: 'org',
  credentialRef: 'signingSecret',       // ce contre quoi chaque livraison est vérifiée
  table: {
    name: 'sensor_reading',
    primaryKey: ['readingId', 'observedAt'],
    timestampColumn: 'observedAt',
    columns: [
      { name: 'readingId', dataType: 'string', isPrimaryKey: true },
      { name: 'deviceId', dataType: 'string' },
      { name: 'value', dataType: 'double', nullable: true },
      { name: 'observedAt', dataType: 'timestamp', isPrimaryKey: true },
    ],
  },
  ingest: {
    mode: 'push',
    deliveryIdHeader: 'x-delivery-id',
    auth: {
      mode: 'hmac',
      headerName: 'x-signature-256',
      algorithm: 'sha256',
      timestampHeader: 'x-signature-timestamp',
    },
    delivery: { itemsPath: '$.readings', format: 'json' },
    mapping: {
      readingId: { path: 'id' },
      deviceId: { path: 'device' },
      value: { path: 'value', transform: 'number_or_null' },
      observedAt: { path: 'ts', transform: 'iso_from_epoch', args: { unit: 'ms' } },
    },
    writeMode: 'upsert',
  },
})

L'émetteur calcule un HMAC-SHA256, avec le secret pour clé, de <horodatage>.<ID de livraison>. suivi des octets exacts qu'il envoie, et l'envoie en hexadécimal minuscule ou majuscule, avec ou sans le préfixe sha256=. Chaque livraison porte son propre ID, et une nouvelle tentative de la même livraison le réutilise avec un nouvel horodatage :

BODY='{"readings":[{"id":"r-1041","device":"pump-7","value":4.2,"ts":1790582820770}]}'
DELIVERY_ID='7d0f3c2e-5b1a-4a8e-9f61-2c4b8e1d0a93'
TS=$(date +%s)
SIG=$(printf '%s.%s.%s' "$TS" "$DELIVERY_ID" "$BODY" | openssl dgst -sha256 -hmac "$SIGNING_SECRET" | awk '{print $NF}')
curl -X POST "https://<push host>/api/platform/data-sources/push/<binding ID>" \
  -H 'Content-Type: application/json' \
  -H "x-delivery-id: $DELIVERY_ID" \
  -H "x-signature-timestamp: $TS" \
  -H "x-signature-256: sha256=$SIG" \
  --data-binary "$BODY"

Les autres vérificateurs ne diffèrent que par auth et par ce que l'émetteur ajoute à sa requête. Chaque onglet est un bloc ingest complet pour la même source :

ingest: {
  mode: 'push',
  deliveryIdHeader: 'x-delivery-id',
  auth: { mode: 'bearer' },
  delivery: { itemsPath: '$.readings', format: 'json' },
  mapping: {
    readingId: { path: 'id' },
    deviceId: { path: 'device' },
    value: { path: 'value', transform: 'number_or_null' },
    observedAt: { path: 'ts', transform: 'iso_from_epoch', args: { unit: 'ms' } },
  },
  writeMode: 'upsert',
},

ingest: {
  mode: 'push',
  deliveryIdHeader: 'x-delivery-id',
  auth: { mode: 'header', headerName: 'x-api-key' },
  delivery: { itemsPath: '$.readings', format: 'json' },
  mapping: {
    readingId: { path: 'id' },
    deviceId: { path: 'device' },
    value: { path: 'value', transform: 'number_or_null' },
    observedAt: { path: 'ts', transform: 'iso_from_epoch', args: { unit: 'ms' } },
  },
  writeMode: 'upsert',
},

ingest: {
  mode: 'push',
  deliveryIdHeader: 'x-delivery-id',
  auth: { mode: 'url_token', paramName: 'token' },
  delivery: { itemsPath: '$.readings', format: 'json' },
  mapping: {
    readingId: { path: 'id' },
    deviceId: { path: 'device' },
    value: { path: 'value', transform: 'number_or_null' },
    observedAt: { path: 'ts', transform: 'iso_from_epoch', args: { unit: 'ms' } },
  },
  writeMode: 'upsert',
},

Une livraison porte au plus 1 Mio et 1 000 lignes, et celle qui dépasse l'une ou l'autre limite est refusée en entier. Découpez un lot plus grand en plusieurs livraisons, chacune avec son propre ID. Ce que l'émetteur reçoit en retour :

Ordre, nouvelles tentatives et livraisons écartées

• Les livraisons sont écrites dans l'ordre où elles ont été acceptées, une à la fois par source. Une livraison n'est écrite qu'une fois que toutes celles acceptées avant elle ont été écrites ou écartées : une livraison plus récente ne passe jamais devant une plus ancienne.
• Une livraison qui ne peut pas être écrite pour l'instant est retentée tant que cela dure — une panne, une table occupée — après 15 secondes, puis après un délai deux fois plus long à chaque fois, au plus 5 minutes d'écart. Pendant qu'elle attend, les livraisons suivantes attendent aussi. Une panne n'écarte jamais une livraison.
• Une livraison que la table refuse pour elle-même 5 fois est écartée, en entier, pour que les suivantes puissent être écrites : des lignes qui ne pourront jamais entrer dans la table, un contenu en conflit avec un lot déjà écrit à la même position, une modification de la source à laquelle ses lignes ne correspondent plus. Pendant une heure, la livraison est conservée avec tout ce qu'elle portait et ce qui lui est arrivé, peut être lue, et peut être remise en fin de file pour être écrite une fois la cause corrigée : adressez-vous à votre opérateur Scrydon. Passé cette heure, elle est supprimée.
• Les lignes d'une livraison qui ne respectent pas les colonnes de la table sont conservées elles aussi, à côté des lignes écrites, pendant la même heure, jamais seulement comptées.
• Une nouvelle tentative avec le même ID de livraison dans l'heure reçoit 202 avec duplicate: true, que la livraison ait été écrite ou écartée, et rien n'est réécrit. Passé cette heure, l'ID est oublié, et une livraison envoyée à nouveau avec cet ID est acceptée comme une nouvelle livraison.

Une source qui ne vérifie rien : auth.mode: 'none'

Une adresse push est dérivée de l'organisation, de la source et de l'environnement : elle n'est pas secrète, et une source qui ne vérifie aucun credential accepterait des lignes de quiconque peut l'atteindre. Une telle source refuse donc chaque livraison tant qu'un administrateur de l'organisation n'a pas ajouté de connexion pour elle — la connexion vaut autorisation :

• Dans l'onglet Connexions de l'extension, la source propose une connexion sans credential, décrite comme Accept unauthenticated pushes from this source. La créer, activée, pour l'environnement de la source accepte les livraisons ; la désactiver ou la supprimer les refuse de nouveau. Les modifications de connexion sont enregistrées dans le journal d'audit avec l'administrateur qui les a effectuées.

• Seule une connexion créée sans authentification vaut autorisation pour la source. Une connexion que l'administrateur crée avec un secret ou un jeton pour les autres sources de l'extension ne la couvre jamais, et une connexion sans authentification ne sert jamais une source qui vérifie un credential. Déclarez le credential de type none de l'extension à côté des autres, pour que le formulaire de connexion puisse le proposer :

  typescript

  auth: {
    credentials: {
      signingSecret: { type: 'apiKey', label: 'Signing secret' },
      none: { type: 'none', label: 'No authentication (accept unsigned deliveries)' },
    },
    default: 'signingSecret',
  },

• D'ici là, la source affiche Needs setup, et les livraisons reçoivent 403 avec push_unauthenticated_not_accepted. Rien n'est mis en file, rien n'est écrit.

• Chaque environnement a besoin de sa propre connexion. Deux connexions activées pour le même environnement refusent les livraisons jusqu'à ce que l'une soit retirée ou que chaque instance soit épinglée à l'une d'elles ; une instance supplémentaire de la source n'accepte que par la connexion à laquelle elle est épinglée.

• Un changement s'applique dès la livraison suivante.

Les sources qui vérifient un credential (bearer, header, hmac, url_token) ne changent pas : leur connexion porte le secret, et une livraison sans connexion utilisable reçoit 503 pour que l'expéditeur réessaie. Ne déclarez none que pour un producteur incapable de s'authentifier.

Passer d'un mode à l'autre

poll et on-demand partagent mot pour mot les mêmes blocs request et response : transformer une source planifiée en source récupérée à la demande revient donc à supprimer intervalSec et writeMode, ajouter params, et conserver tout le reste. L'inverse est vrai également.

État des connexions

Les sources installées affichent l’état de leur connexion pour l’environnement d’espace de travail sélectionné, dans Analytics et dans les paramètres d’extension de Platform. Les sources manuelles et à lignes statiques sont intégrées. Les autres modes déduisent leurs besoins en credential de leur référence d’authentification déclarée ; un credential partagé est compté une fois pour chaque source qui l’utilise. « Ready » confirme les métadonnées du credential applicable, pas une requête réussie auprès du fournisseur.

Utilisez Configure pour ouvrir le panneau Connections existant de l’extension. Après la configuration d’un credential, revenir dans Analytics actualise l’état. Une configuration manquante peut masquer une source installée dans la liste Data Sources ; un service d’état indisponible ne masque pas les sources, et la page Extensions reste disponible pour la découverte. Les sources query exigent toujours une connexion basic-auth activée dans l’environnement exact.

Historique opérationnel des synchronisations

Le panneau de détail affiche les 50 derniers événements et propose un lien vers l’historique de la table lorsqu’elle existe. Les lots stream réussis sont regroupés par minute UTC, table, opération et acteur, avec le nombre de lots et de lignes validés. Rejouer un reçu déjà validé n’augmente pas ces compteurs. Les erreurs et les événements des autres modes restent individuels.

La maintenance horaire conserve les 1 000 derniers événements par source, organisation et environnement, ainsi que toutes les erreurs de moins de 30 jours. Chaque passage supprime au maximum 1 000 événements éligibles plus anciens ; un retard peut donc nécessiter plusieurs passages. Cette politique est distincte de la rétention des données et des enregistrements individuels d’audit de conformité.

Lire l'état et l'historique d'une source

Last sync succeeded décrit le dernier résultat d’acquisition enregistré ; il ne confirme pas qu’un stream est actuellement connecté. Data stale s’affiche séparément lorsque la table sous-jacente a dépassé son seuil de fraîcheur. Les états en pause et en échec restent visibles indépendamment de la fraîcheur de la table.

Les nouveaux lots stream réussis apparaissent sous forme de résumés par minute avec le nombre de lots et de lignes validés. Les anciens événements individuels et les erreurs restent individuels ; l’historique n’est pas réécrit. Un événement sans durée mesurée n’affiche pas de durée.

Le stockage est plein : libérez de l’espace ou ajoutez de la capacité. s’affiche (enregistré comme storage_full) lorsque le stockage du déploiement n’a plus d’espace disque, si bien qu’une écriture ou une nouvelle table a été refusée. Rien de cette synchronisation n’a été écrit et rien n’est perdu : un lot stream ou une livraison push est réessayé, et l’exécution planifiée suivante réessaie. Cela ne rapproche jamais une source de la mise en pause automatique. Votre opérateur Scrydon corrige le problème en libérant de l’espace ou en ajoutant de la capacité ; les sources reprennent ensuite d’elles-mêmes.

Votre politique DLP s'applique à chaque ligne

Avant qu'une ligne soit écrite dans la table d'une source — dans tous les modes, miroirs query compris — la plateforme l'analyse avec la politique DLP de votre organisation (Settings → Governance → DLP), la même qui gouverne les appels de capacités. Chaque ligne est décidée seule :

• Une ligne que la politique réécrit (masquage, caviardage, hachage) est stockée réécrite. La réponse d'un appelant on-demand est faite des lignes stockées : la même réécriture, sans les lignes écartées, que son résultat de persistance compte (persistence.dataPolicy). Fetch now dans le détail d'une source indique combien : 2 rows set aside by the data policy.
• En mode Enforce, une ligne qui présente une détection est mise de côté et jamais stockée. Les lignes voisines sont écrites normalement.
• Une ligne seule plus grande qu'une analyse (1 Mio) ne peut pas être vérifiée : elle est donc mise de côté et jamais stockée, dans tous les modes. Les lignes voisines sont analysées et écrites normalement.
• Rien à faire : la ligne est stockée telle qu'elle est arrivée.

L'état de la source indique ce que la politique a fait : data_policy_applied: The organization's data policy redacted 2 rows and set aside 1 rows it blocks (LLM_BLOCKED_BY_DLP) before they were written. Une ligne mise de côté parce que trop grande y est nommée aussi : 1 rows were set aside unwritten, each larger than the organization's data policy scans at once (1 MiB). C'est un avertissement à côté d'une synchronisation réussie, jamais un échec : il ne rapproche jamais une source de la pause automatique. Les détections sont consignées dans le journal d'audit sous la source, sans les valeurs détectées.

Si la politique ne peut pas être vérifiée, rien n'est écrit et rien n'est perdu : un lot stream ou une livraison push est réessayé, un import manuel ou une requête on-demand signale un échec que vous pouvez relancer, et une exécution poll planifiée est enregistrée comme data_policy_unavailable, l'exécution suivante servant de nouvel essai. Un nouvel essai n'écrit jamais un lot deux fois.