Merge pull request #32 from netlify/fix-rate-limit

Fix rate limit

Author: bcomnes

README.md

@@ -51,9 +51,9 @@ Every open-api method has the following signature:
 
 Perform a call to the given endpoint corresponding with the `operationId`. Returns promise that will resolve with the body of the response, or reject with an error with details about the request attached. Rejects if the `status` > 400. Successful response objects have `status` and `statusText` properties on their prototype.
 
-- `params` is an object that includes any of the required or optional endpoint parameters.  
-- `params.body` should be an object which gets serialized to JSON automatically.  
-- If the endpoint accepts `binary`, `params.body` can be a Node.js readable stream.
+- `params` is an object that includes any of the required or optional endpoint parameters.
+- `params.body` should be an object which gets serialized to JSON automatically.
+- If the endpoint accepts `binary`, `params.body` can be a Node.js readable stream or stream ctor (e.g. `() => fs.createReadStream('./foo')`).  A Stream ctor function is required to support rate limit retry.
 
 ```js
 // example params
@@ -125,7 +125,7 @@ async function login () {
   // Open browser for authentication
   await openBrowser(`https://app.netlify.com/authorize?response_type=ticket&ticket=${ticket.id}`)
   const accessToken = await api.getAccessToken(ticket)
-  // API is also set up to use the returned access token as a side effect 
+  // API is also set up to use the returned access token as a side effect
   return accessToken // Save this for later so you can quickly set up an authenticated client
 }
 ```

package.json

@@ -43,6 +43,7 @@
     "folder-walker": "^3.2.0",
     "from2-array": "0.0.4",
     "hasha": "^3.0.0",
+    "is-stream": "^1.1.0",
     "lodash.camelcase": "^4.3.0",
     "lodash.flatten": "^4.4.0",
     "lodash.get": "^4.4.2",

src/deploy/hash-fns.js

@@ -14,7 +14,7 @@ async function hashFns(dir, opts) {
       assetType: 'function',
       hashAlgorithm: 'sha256',
       // tmpDir,
-      statusCb: () => {}
+      statusCb: () => { }
     },
     opts
   )
@@ -36,7 +36,7 @@ async function hashFns(dir, opts) {
   const manifestCollector = manifestCollectorCtor(functions, fnShaMap, opts)
 
   // TODO: Zip up functions, hash then send.
-  // This is totally wrong wright now.
+  // This is totally wrong right now.
   // See https://github.com/netlify/open-api/blob/master/go/porcelain/deploy.go#L544
 
   await pump(fileStream, fnStat, fnFilter, hasher, manifestCollector)

src/deploy/upload-files.js

@@ -13,7 +13,7 @@ async function uploadFiles(api, deployId, uploadList, { concurrentUpload, status
 
   const uploadFile = async (fileObj, index) => {
     const { normalizedPath, assetType, runtime } = fileObj
-    const readStream = fs.createReadStream(fileObj.filepath)
+    const readStreamCtor = () => fs.createReadStream(fileObj.filepath)
 
     statusCb({
       type: 'upload',
@@ -26,7 +26,7 @@ async function uploadFiles(api, deployId, uploadList, { concurrentUpload, status
         response = await retryUpload(
           () =>
             api.uploadDeployFile({
-              body: readStream,
+              body: readStreamCtor,
               deployId,
               path: encodeURI(normalizedPath)
             }),
@@ -38,7 +38,7 @@ async function uploadFiles(api, deployId, uploadList, { concurrentUpload, status
         response = await await retryUpload(
           () =>
             api.uploadDeployFunction({
-              body: readStream,
+              body: readStreamCtor,
               deployId,
               name: encodeURI(normalizedPath),
               runtime

src/index.test.js

@@ -5,7 +5,7 @@ const NetlifyAPI = require('./index')
 const body = promisify(require('body'))
 const fromString = require('from2-string')
 const { TextHTTPError } = require('micro-api-client')
-const { existy } = require('./open-api/util')
+const { existy, unixNow } = require('./open-api/util')
 
 const createServer = handler => {
   const s = http.createServer(handler)
@@ -252,11 +252,12 @@ test.serial('test rate-limiting', async t => {
   try {
     server = createServer(async (req, res) => {
       if (!existy(retryAt)) {
-        retryAt = Date.now() + randomInt(1000, 5000) //ms
+        retryAt = unixNow() + randomInt(1, 5) //ms
 
         requestRateLimit(res, retryAt)
       } else {
-        const rateLimitFinished = Date.now() >= retryAt
+        const now = unixNow()
+        const rateLimitFinished = now >= retryAt
 
         if (rateLimitFinished) {
           t.pass('The client made a request at or after the rate limit deadline')

src/open-api/index.js

@@ -7,7 +7,8 @@ const Headers = fetch.Headers
 const camelCase = require('lodash.camelcase')
 const { JSONHTTPError, TextHTTPError } = require('micro-api-client')
 const debug = require('debug')('netlify:open-api')
-const { existy, sleep } = require('./util')
+const { existy, sleep, unixNow } = require('./util')
+const isStream = require('is-stream')
 
 exports.methods = require('./shape-swagger')
 
@@ -92,7 +93,15 @@ exports.generateMethod = method => {
       let response
       try {
         debug(`requesting ${path}`)
-        response = await fetch(path, opts)
+
+        if (isStream(opts.body) && opts.body.closed) {
+          throw new Error('Body readable stream is already used.  Try passing a stream ctor instead')
+        }
+
+        // Pass in a function for readable stream ctors
+        const fetchOpts = typeof opts.body === 'function' ? Object.assign({}, opts, { body: opts.body() }) : opts
+
+        response = await fetch(path, fetchOpts)
       } catch (e) {
         // TODO: clean up this error path
         debug(`fetch error`)
@@ -128,7 +137,9 @@ exports.generateMethod = method => {
               throw new Error('Header missing reset time')
             }
             debug(`resetTime: ${resetTime}`)
-            const sleepTime = resetTime - Date.now()
+            const now = unixNow()
+            debug(`unixNow(): ${now}`)
+            const sleepTime = (resetTime - now < 0 ? 0 : resetTime - now) * 1000
             debug(`sleeping for ${sleepTime}ms`)
             await sleep(sleepTime)
           } catch (e) {

src/open-api/util.js

@@ -34,3 +34,5 @@ exports.existy = val => val != null
 
 // Async sleep.  A whole new WORLD!
 exports.sleep = ms => new Promise(resolve => setTimeout(resolve, ms))
+
+exports.unixNow = () => Math.floor(new Date() / 1000)