AWS Lambda自定义运行时流式响应报错RuntimeResponseSentState求助

尝试在AWS Lambda自定义Runtime中实现响应流式传输，按照官方文档操作，但返回第二个数据块时触发RuntimeResponseSentState错误。同时想确认AWS自定义Runtime API是基于HTTP请求还是Socket连接。

我的Golang代码 #

func main() {
    clientTrace := &httptrace.ClientTrace{
        GotConn: func(info httptrace.GotConnInfo) { log.Printf("conn was reused: %t", info.Reused) },
    }
    endpoint := os.Getenv("AWS_LAMBDA_RUNTIME_API")
    traceCtx := httptrace.WithClientTrace(context.Background(), clientTrace)

    // 1st request
    req1, err := http.NewRequestWithContext(traceCtx, http.MethodGet, "http://"+endpoint+"/2018-06-01/runtime/invocation/next", nil)
    if err != nil {
        log.Fatal(err)
    }
    res1, err := http.DefaultClient.Do(req1)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(res1)
    reqId := res1.Header.Get("Lambda-Runtime-Aws-Request-Id")
    os.Setenv("_X_AMZN_TRACE_ID", res1.Header.Get("Lambda-Runtime-Trace-Id"))

    b, err := io.ReadAll(res1.Body)
    fmt.Println(string(b))
    if _, err := io.Copy(ioutil.Discard, res1.Body); err != nil {
        log.Fatal(err)
    }
    res1.Body.Close()
    // 2nd request
    req2, err := http.NewRequestWithContext(traceCtx, http.MethodPost, "http://"+endpoint+"/2018-06-01/runtime/invocation/"+reqId+"/response", bytes.NewBuffer([]byte("data : Hello")))
    if err != nil {
        log.Fatal(err)
    }
    req2.Header.Set("_X_AMZN_TRACE_ID", res1.Header.Get("Lambda-Runtime-Trace-Id"))
    req2.Header.Set("Lambda-Runtime-Function-Response-Mode", "streaming")
    req2.Header.Set("Transfer-Encoding", "chunked")
    req2.TransferEncoding = []string{"chunked"}
    res2, err := http.DefaultClient.Do(req2)
    fmt.Println(res2)
    if err != nil {
        log.Fatal(err)
    }
    if _, err := io.Copy(ioutil.Discard, res2.Body); err != nil {
        log.Fatal(err)
    }
    b, _ = io.ReadAll(res2.Body)
    fmt.Println(string(b))
    res2.Body.Close()
    req3, err := http.NewRequestWithContext(traceCtx, http.MethodPost, "http://"+endpoint+"/2018-06-01/runtime/invocation/"+reqId+"/response", bytes.NewBuffer([]byte("data : Hello")))
    if err != nil {
        log.Fatal(err)
    }
    req3.Header.Set("_X_AMZN_TRACE_ID", res1.Header.Get("Lambda-Runtime-Trace-Id"))
    req3.Header.Set("Lambda-Runtime-Function-Response-Mode", "streaming")
    req3.TransferEncoding = []string{"chunked"}
    res3, err := http.DefaultClient.Do(req3)
    fmt.Println(res3)
    b, _ = io.ReadAll(res3.Body)
    fmt.Println(string(b))
    if err != nil {
        log.Fatal(err)
    }
    if _, err := io.Copy(ioutil.Discard, res3.Body); err != nil {
        log.Fatal(err)
    }
    res3.Body.Close()
}

AWS控制台报错信息 #

2023/11/21 06:58:58 conn was reused: false
START RequestId: 789ebfcf-9a1b-4517-a9d5-7d408c4c1c45 Version: $LATEST
&{200 OK 200 HTTP/1.1 1 1 map[Content-Length:[49] Content-Type:[application/json] Date:[Tue, 21 Nov 2023 06:58:58 GMT] Lambda-Runtime-Aws-Request-Id:[789ebfcf-9a1b-4517-a9d5-7d408c4c1c45] Lambda-Runtime-Deadline-Ms:[1700549945351] Lambda-Runtime-Invoked-Function-Arn:[arn:aws:lambda:us-east-1:130250231763:function:myFunction2] Lambda-Runtime-Trace-Id:[Root=1-655c5532-4acef3f33a57d61a7c4bcc0c;Parent=515fefae0cd6bee6;Sampled=0;Lineage=f35aca94:0]] 0xc000076300 49 [] false false map[] 0xc0000a8000 <nil>}
{
  "key1": "value1",
  "key2": "value2",
  "key3": "value3"
}
2023/11/21 06:58:58 conn was reused: true
&{202 Accepted 202 HTTP/1.1 1 1 map[Content-Length:[16] Content-Type:[application/json] Date:[Tue, 21 Nov 2023 06:58:58 GMT]] 0xc00010a080 16 [] false false map[] 0xc0000a8200 <nil>}
2023/11/21 06:58:58 conn was reused: true
&{403 Forbidden 403 HTTP/1.1 1 1 map[Content-Length:[184] Content-Type:[application/json] Date:[Tue, 21 Nov 2023 06:58:58 GMT]] 0xc00010a140 184 [] false false map[] 0xc000110200 <nil>}
{
  "errorMessage": "State transition from RuntimeResponseSentState to InvocationResponse failed for runtime. Error: State transition is not allowed",
  "errorType": "InvalidStateTransition"
}
RequestId: 789ebfcf-9a1b-4517-a9d5-7d408c4c1c45 Error: Runtime exited without providing a reason
Runtime.ExitError

问题解答 #

1. 报错原因及修复方法 #

• 错误原因：对同一个请求ID发起了两次独立的POST /response请求，不符合流式响应规则。流式响应要求在同一个HTTP连接中通过分块编码（chunked）发送所有数据块，而非多次调用/response端点。第一次请求返回202 Accepted后，Runtime已进入响应发送状态，不允许再对该请求ID发起第二次响应请求，因此触发状态转换错误。

• 修复后的代码示例：

import (
    "context"
    "fmt"
    "io"
    "log"
    "net/http"
    "net/http/httptrace"
    "os"
    "time"
)

func main() {
    clientTrace := &httptrace.ClientTrace{
        GotConn: func(info httptrace.GotConnInfo) { log.Printf("conn was reused: %t", info.Reused) },
    }
    endpoint := os.Getenv("AWS_LAMBDA_RUNTIME_API")
    traceCtx := httptrace.WithClientTrace(context.Background(), clientTrace)

    // 获取下一个调用请求
    req1, err := http.NewRequestWithContext(traceCtx, http.MethodGet, "http://"+endpoint+"/2018-06-01/runtime/invocation/next", nil)
    if err != nil {
        log.Fatal(err)
    }
    res1, err := http.DefaultClient.Do(req1)
    if err != nil {
        log.Fatal(err)
    }
    defer res1.Body.Close()

    reqId := res1.Header.Get("Lambda-Runtime-Aws-Request-Id")
    traceId := res1.Header.Get("Lambda-Runtime-Trace-Id")
    os.Setenv("_X_AMZN_TRACE_ID", traceId)

    // 读取请求体
    b, err := io.ReadAll(res1.Body)
    fmt.Println(string(b))
    if err != nil {
        log.Fatal(err)
    }

    // 创建流式响应请求，使用分块编码
    req2, err := http.NewRequestWithContext(traceCtx, http.MethodPost, "http://"+endpoint+"/2018-06-01/runtime/invocation/"+reqId+"/response", nil)
    if err != nil {
        log.Fatal(err)
    }
    req2.Header.Set("_X_AMZN_TRACE_ID", traceId)
    req2.Header.Set("Lambda-Runtime-Function-Response-Mode", "streaming")
    req2.TransferEncoding = []string{"chunked"}

    // 获取请求的body writer，用于分块写入数据
    client := &http.Client{}
    resp, err := client.Do(req2)
    if err != nil {
        log.Fatal(err)
    }
    defer resp.Body.Close()

    // 获取响应的writer，写入第一个数据块
    writer := req2.Body.(io.Writer)
    _, err = writer.Write([]byte("data : Hello\n"))
    if err != nil {
        log.Fatal(err)
    }
    // 刷新缓冲区，确保数据发送
    if flusher, ok := writer.(http.Flusher); ok {
        flusher.Flush()
    }

    // 模拟延迟，写入第二个数据块
    time.Sleep(1 * time.Second)
    _, err = writer.Write([]byte("data : Second chunk\n"))
    if err != nil {
        log.Fatal(err)
    }
    if flusher, ok := writer.(http.Flusher); ok {
        flusher.Flush()
    }

    // 关闭writer，结束响应
    req2.Body.(io.Closer).Close()

    // 读取Runtime的响应
    respBody, err := io.ReadAll(resp.Body)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(string(respBody))
}

• 关键修改点：
  • 仅创建一个POST /response请求，保持连接打开，通过同一个writer分块写入多个数据块
  • 使用http.Flusher强制刷新缓冲区，确保数据及时发送到Runtime
  • 所有数据发送完成后再关闭请求的body，结束响应

2. AWS自定义Runtime API类型确认 #

AWS自定义Runtime API是基于HTTP/1.1协议的，通过标准HTTP请求与Lambda Runtime API端点交互，并非Socket连接。可通过环境变量AWS_LAMBDA_RUNTIME_API获取端点地址，发送GET/POST等HTTP请求完成调用获取、响应发送、错误上报等操作。

内容的提问来源于stack exchange，提问作者puntsag zorigoo