冲突检测和同步 - AWS AppSync
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 AWS 服务入门

冲突检测和同步

版本化数据源

AWS AppSync 目前支持 DynamoDB 数据源的版本控制。冲突检测、冲突解决和同步操作需要 Versioned 数据源。对数据源启用版本控制时,AWS AppSync 将自动执行以下操作:

  • 使用对象版本化元数据增强项目。

  • 将使用 AWS AppSync 更改对项目所做的更改记录到增量 表。

  • 使用“逻辑删除”将基本 表中的已删除项目保留可配置的时间。

版本化数据源配置

对 DynamoDB 数据源启用版本控制时,可以指定以下字段:

BaseTableTTL

使用“逻辑删除”(一个元数据字段,指示项目已被删除)保留基本 表中已删除项目的分钟数。如果希望删除项目时立即将其移除,则可以将此值设置为 0。此字段为必填。

DeltaSyncTableName

表的名称,用于存储使用 AWS AppSync 更改对项目所做的更改。此字段为必填。

DeltaSyncTableTTL

增量 表中保留项目的分钟数。此字段为必填。

增量同步表

AWS AppSync 目前支持使用 PutItemUpdateItemDeleteItem DynamoDB 操作对更改进行增量同步日志记录。

当 AWS AppSync 更改将更改版本化数据源中的项目时,该更改的记录将存储在针对增量更新优化的增量 表中。您可以选择对其他版本化数据源使用不同的增量 表(例如,每种类型一个,每个域区域一个),或者对 API 使用单个增量 表。AWS AppSync 建议不要对多个 API 使用单个增量 表,以避免主键冲突。

此表所需的架构如下所示:

ds_pk

用作分区键的字符串值。它是通过将基本 数据源名称和 ISO8601 格式的更改发生日期连接起来而构建的。(例如 Comments:2019-01-01)

ds_sk

用作排序键的字符串值。它是通过连接 IS08601 格式的更改发生时间、项目的主键和项目的版本来构建的。这些字段的组合保证了增量表中每个条目的唯一性(例如,对于时间 09:30:00、ID 1a、版本 2,该值将为 09:30:00:1a:2)

_ttl

一个数值,用于存储应从增量 表中移除项目时的时间戳(以纪元秒为单位)。此值是通过将数据源上配置的 DeltaSyncTableTTL 值添加到发生更改那一刻的时间来确定的。此字段应配置为 DynamoDB TTL 属性。

配置为与基本 表一起使用的 IAM 角色还必须包含对增量 表进行操作的权限。在此示例中,显示了名为 Comments基本 表和名为 ChangeLog增量 表的权限策略:

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "dynamodb:DeleteItem", "dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:Query", "dynamodb:Scan", "dynamodb:UpdateItem" ], "Resource": [ "arn:aws:dynamodb:us-east-1:000000000000:table/Comments", "arn:aws:dynamodb:us-east-1:000000000000:table/Comments/*", "arn:aws:dynamodb:us-east-1:000000000000:table/ChangeLog", "arn:aws:dynamodb:us-east-1:000000000000:table/ChangeLog/*" ] } ] }

版本化数据源元数据

AWS AppSync 代表您管理 Versioned 数据源上的元数据字段。自行修改这些字段可能会导致应用程序中的错误或数据丢失。这些字段包括:

_version

一个单调递增的计数器,在项目发生更改时随时更新。

_lastChangedAt

一个数值,用于存储上次修改项目时的时间戳(以纪元毫秒为单位)。

_deleted

一个布尔型“逻辑删除”值,指示项目已被删除。应用程序可以使用此功能从本地数据存储中移除已删除的项目。

_ttl

一个数值,用于存储应从底层数据源中移除项目时的时间戳(以纪元秒为单位)。

ds_pk

用作增量 表的分区键的字符串值。

ds_sk

用作增量 表的排序键的字符串值。

这些元数据字段将影响底层数据源中项目的总体大小。AWS AppSync 建议在设计应用程序时为版本化数据源元数据保留 500 字节 + 最大主键大小 的存储空间。要在客户端应用程序中使用此元数据,请在您的 GraphQL 类型和更改的选择集中包括 _version_lastChangedAt_deleted 字段。

冲突检测和解决

当 AWS AppSync 发生并发写入时,您可以配置冲突检测和冲突解决策略以适当处理更新。冲突检测确定更改是否与数据源中的实际写入项目发生冲突。通过将 conflictDetection 字段的 SyncConfig 中的值设置为 VERSION 来启用冲突检测。

冲突解决是在检测到冲突时执行的操作。这是通过在 SyncConfig 中设置冲突处理程序字段来确定的。有三种冲突解决策略:

  • OPTIMISTIC_CONCURRENCY

  • AUTOMERGE

  • LAMBDA

下文分别详细介绍了这些解决冲突策略。注意: 无法在 PIPELINE 解析程序上启用同步。

在写入操作期间,版本将由 AppSync 自动递增,而不应由客户端或在配置了版本化数据源的解析程序之外进行修改。否则会改变系统的一致性行为,并可能导致数据丢失。

乐观并发

乐观并发是 AWS AppSync 为版本化数据源提供一种冲突解决策略。当冲突解决程序设置为乐观并发时,如果检测到传入的更改具有与对象的实际版本不同的版本,则冲突处理程序将简单地拒绝传入请求。在 GraphQL 响应中,将提供服务器上具有最新版本的现有项目。然后,客户端需要在本地处理此冲突,并使用项目的更新版本重试更改。

Automerge

Automerge 为开发人员提供了配置冲突解决策略的简单方法,而无需编写客户端逻辑来手动合并其他策略无法处理的冲突。Automerge 在合并数据以解决冲突时遵守严格的规则集。Automerge 的原则围绕 GraphQL 字段的底层数据类型。这些原则如下所示:

  • 标量字段上的冲突:GraphQL 标量或不是集合的任何字段(即列表、集合、映射)。拒绝标量字段的传入值并选择服务器中现有的值。

  • 列表上的冲突:GraphQL 类型和数据库类型是列表。将传入列表与服务器中的现有列表连接起来。传入更改中的列表值将附加到服务器中列表的末尾。将保留重复的值。

  • 集合上的冲突:GraphQL 类型是一个列表,数据库类型是一个集合。使用传入集合和服务器中的现有集合应用集合合并。这符合集合的属性,意味着没有重复的条目。

  • 当传入的更改向项目添加新字段时,应将其合并到现有项目中。

  • 映射上的冲突:当数据库中的底层数据类型是映射(即键值文档)时,在解析和处理映射的每个属性时应用上述规则。

Automerge 旨在使用更新版本自动检测、合并和重试请求,从而使客户端无需手动合并任何冲突的数据。

为了显示 Automerge 如何处理标量类型上的冲突的示例,我们将使用以下记录作为起点。

{ "id" : 1, "name" : "Nadia", "jersey" : 5, "_version" : 4 }

现在,传入的更改可能正在尝试更新该项目,但使用的是较旧版本,因为客户端尚未与服务器同步。它类似于以下内容:

{ "id" : 1, "name" : "Nadia", "jersey" : 55, "_version" : 2 }

请注意传入请求中的过时版本 2。在此流程中,Automerge 将通过拒绝“球衣”字段更新为“55”来合并数据,并将其值保持在“5”,从而导致下面的项目图像保存在服务器中。

{ "id" : 1, "name" : "Nadia", "jersey" : 5, "_version" : 5 # version is incremented every time automerge performs a merge that is stored on the server. }

鉴于上面显示的具有版本 5 的项目状态,现在假设一个传入的更改尝试使用以下图像改变项目:

{ "id" : 1, "name" : "Shaggy", "jersey" : 5, "interests" : ["breakfast", "lunch", "dinner"] # underlying data type is a Set "points": [24, 30, 27] # underlying data type is a List "_version" : 3 }

传入的更改中有三个兴趣点。名称(一个标量)已被更改,但添加了两个新字段“兴趣”(集合)和“分数”(列表)。在这种情况下,将检测到由于版本不匹配而导致的冲突。Automerge 将遵循其属性并拒绝名称更改,因为它是一个标量并添加到非冲突字段。这将导致保存在服务器中的项目显示如下。

{ "id" : 1, "name" : "Nadia", "jersey" : 5, "interests" : ["breakfast", "lunch", "dinner"] # underlying data type is a Set "points": [24, 30, 27] # underlying data type is a List "_version" : 6 }

使用具有版本 6 的项目的更新图像,现在假设传入的更改(版本不匹配)尝试将项目转换为以下内容:

{ "id" : 1, "name" : "Nadia", "jersey" : 5, "interests" : ["breakfast", "lunch", "brunch"] # underlying data type is a Set "points": [30, 35] # underlying data type is a List "_version" : 5 }

在这里,我们观察到“兴趣”的传入字段有一个存在于服务器中的重复值和两个新值。在这种情况下,由于底层数据类型是集合,Automerge 会将服务器中现有的值与传入请求中的值合并,并剔除任何重复项。同样,“分数”字段上也存在冲突,其中有一个重复值和一个新值。但是,由于这里的底层数据类型是列表,Automerge 会简单地将传入请求中的所有值附加到服务器中已存在的值的末尾。存储在服务器上的结果合并图像如下所示:

{ "id" : 1, "name" : "Nadia", "jersey" : 5, "interests" : ["breakfast", "lunch", "dinner", "brunch"] # underlying data type is a Set "points": [24, 30, 27, 30, 35] # underlying data type is a List "_version" : 7 }

现在,让我们假设存储在服务器中的项目(版本 8)显示如下。

{ "id" : 1, "name" : "Nadia", "jersey" : 5, "interests" : ["breakfast", "lunch", "dinner", "brunch"] # underlying data type is a Set "points": [24, 30, 27, 30, 35] # underlying data type is a List "stats": { "ppg": "35.4", "apg": "6.3" } "_version" : 8 }

但是传入的请求尝试使用以下图像更新该项目,从而再次出现版本不匹配的情况:

{ "id" : 1, "name" : "Nadia", "stats": { "ppg": "25.7", "rpg": "6.9" } "_version" : 3 }

现在,在这种情况下,我们可以看到服务器中已经存在的字段丢失(兴趣、分数、球衣)。此外,正在编辑映射“统计”中的“ppg”值,添加了一个新值“rpg”,并省略了“apg”。Automerge 保留已省略的字段(注意:如果打算删除字段,则必须使用匹配版本再次尝试请求),以便它们不会丢失。它也会将相同的规则应用于映射中的字段,因此对“ppg”的更改将被拒绝,而“apg”被保留,并添加一个新字段“rpg”。存储在服务器中的结果项目现在将显示为:

{ "id" : 1, "name" : "Nadia", "jersey" : 5, "interests" : ["breakfast", "lunch", "dinner", "brunch"] # underlying data type is a Set "points": [24, 30, 27, 30, 35] # underlying data type is a List "stats": { "ppg": "35.4", "apg": "6.3", "rpg": "6.9" } "_version" : 9 }

Lambda

冲突解决选项:

  • RESOLVE:将现有项替换为响应有效负载中提供的新项。您一次只能对单个项目重试同一操作。当前受 DynamoDB PutItemUpdateItem 支持。

  • REJECT:拒绝更改并返回错误以及 GraphQL 响应中的现有项目。当前受 DynamoDB PutItemUpdateItemDeleteItem 支持。

  • REMOVE:删除现有项目。当前受 DynamoDB DeleteItem 支持。

Lambda 调用请求

AWS AppSync DynamoDB 解析程序将调用在 LambdaConflictHandlerArn 中指定的 Lambda 函数。它将使用在数据源上配置的相同 service-role-arn。调用的负载具有以下结构:

{ "newItem": { ... }, "existingItem": {... }, "arguments": { ... }, "resolver": { ... }, "identity": { ... } }

字段定义如下:

newItem

预览项目(如果更改成功)。

existingItem

该项目当前位于 DynamoDB 表中。

arguments

来自 GraphQL 更改的参数。

resolver

有关 AWS AppSync 解析程序的信息。

identity

有关调用方的信息。如果使用 API 密钥进行访问,则此字段设置为 null。

有效负载示例:

{ "newItem": { "id": "1", "author": "Jeff", "title": "Foo Bar", "rating": 5, "comments": ["hello world"], }, "existingItem": { "id": "1", "author": "Foo", "rating": 5, "comments": ["old comment"] }, "arguments": { "id": "1", "author": "Jeff", "title": "Foo Bar", "comments": ["hello world"] }, "resolver": { "tableName": "post-table", "awsRegion": "us-west-2", "parentType": "Mutation", "field": "updatePost" }, "identity": { "accountId": "123456789012", "sourceIp": "x.x.x.x", "username": "AIDAAAAAAAAAAAAAAAAAA", "userArn": "arn:aws:iam::123456789012:user/appsync" } }

Lambda 调用响应

对于 PutItemUpdateItem 解决冲突

RESOLVE 更改。响应必须采用以下格式。

{ "action": "RESOLVE", "item": { ... } }

item 字段表示将用于替换底层数据源中的现有项目的对象。如果在 item 中包含主键和同步元数据,则将被忽略。

REJECT 更改。响应必须采用以下格式。

{ "action": "REJECT" }

对于 DeleteItem 冲突解决

REMOVE 项目。响应必须采用以下格式。

{ "action": "REMOVE" }

REJECT 更改。响应必须采用以下格式。

{ "action": "REJECT" }

下面的示例 Lambda 函数检查谁进行调用以及解析程序名称。如果由 jeffTheAdmin 调用,则 REMOVE DeletePost 解析程序的对象,或 RESOLVE 与更新/放置解析程序的新项目冲突。如果不是,则更改是 REJECT

exports.handler = async (event, context, callback) => { console.log("Event: "+ JSON.stringify(event)); // Business logic goes here. var response; if ( event.identity.user == "jeffTheAdmin" ) { let resolver = event.resolver.field; switch(resolver) { case "deletePost": response = { "action" : "REMOVE" } break; case "updatePost": case "createPost": response = { "action" : "RESOLVE", "item": event.newItem } break; default: response = { "action" : "REJECT" }; } } else { response = { "action" : "REJECT" }; } console.log("Response: "+ JSON.stringify(response)); return response; }

错误

ConflictUnhandled

冲突检测发现版本不匹配情况,并且冲突处理程序拒绝更改。

示例:使用乐观并发冲突处理程序解决冲突。或者,Lambda 冲突处理程序返回 REJECT

ConflictError

尝试解决冲突时发生内部错误。

示例:Lambda 冲突处理程序返回格式错误的响应。或者,无法调用 Lambda 冲突处理程序,因为找不到提供的资源 LambdaConflictHandlerArn

MaxConflicts

已达到解决冲突的最大重试次数。

示例:同一对象上的并发请求太多。在解决冲突之前,另一个客户端将对象更新为新版本。

BadRequest

客户端尝试更新元数据字段(_version_ttl_lastChangedAt_deleted)。

示例:客户端尝试使用更新更改来更新对象的版本。

DeltaSyncWriteError

写入增量同步记录失败。

示例:更改成功,但尝试写入增量同步表时出现内部错误。

InternalFailure

出现内部错误。

CloudWatch Logs (CloudWatch 日志)

如果 AWS AppSync API 已启用 CloudWatch Logs,并将日志记录设置“字段级日志”设为 enabled,将“字段级日志”的日志级别设为 ALL,则 AWS AppSync 会向日志组发送冲突检测和解决信息。有关日志消息格式的信息,请参阅冲突检测和同步日志记录的文档

同步操作

版本化数据源支持 Sync 操作,以允许您从 DynamoDB 表中检索所有结果,然后仅接收自上次查询(增量更新)以来更改的数据。当 AWS AppSync 收到 Sync 操作请求时,它会使用请求中指定的字段来确定是否应访问基本 表或增量 表。

  • 如果未指定 lastSync 字段,则会对基本 表执行 Scan

  • 如果指定了 lastSync 字段,但值在 current moment - DeltaSyncTTL 之前,则会对基本 表执行 Scan

  • 如果指定了 lastSync 字段,并且值在 current moment - DeltaSyncTTL 或之后,则会对增量 表执行 Query

AWS AppSync 将 startedAt 字段返回到所有 Sync 操作的响应映射模板。startedAt 字段是 Sync 操作开始的时刻,以纪元毫秒为单位,您可以在本地存储并在其他请求中使用该值。如果请求中包含分页令牌,则该值将与请求针对第一页结果返回的值相同。

有关 Sync 映射模板格式的信息,请参阅映射模板参考