etl: add timeout option for individual object transform request

Nahemah1022 · Nahemah1022 · commit 25513e7c047f · 2025-05-14T09:02:51.000-07:00
* previously, `msg.Timeout` was used for both etl initialization and
  individual object transform requests.
* this patch introduce `msg.ObjTimeout` to allow separate config for
  per-object transform request timeout.
* TODO: update CLI with this change.

Signed-off-by: Tony Chen &lt;a122774007@gmail.com&gt;
diff --git a/ais/test/etl_stress_test.go b/ais/test/etl_stress_test.go
@@ -60,7 +60,7 @@ def transform(input_bytes):
 	m.puts()
 
 	msg := etl.InitCodeMsg{
-		InitMsgBase: etl.InitMsgBase{EtlName: "etl-build-conn-err", Timeout: etlBucketTimeout},
+		InitMsgBase: etl.InitMsgBase{EtlName: "etl-build-conn-err", InitTimeout: etlBucketTimeout},
 		Code:        []byte(timeoutFunc),
 		Runtime:     runtime.Py39,
 		ChunkSize:   0,
@@ -218,7 +218,7 @@ def transform(input_bytes):
 				etlName = test.name
 				{
 					test.etlCodeMsg.EtlName = etlName
-					test.etlCodeMsg.Timeout = etlBucketTimeout
+					test.etlCodeMsg.InitTimeout = etlBucketTimeout
 					test.etlCodeMsg.Funcs.Transform = "transform"
 				}
 				_ = tetl.InitCode(t, baseParams, &test.etlCodeMsg)
diff --git a/ais/test/etl_test.go b/ais/test/etl_test.go
@@ -785,9 +785,9 @@ def transform(input_bytes: bytes) -> bytes:
 			t.Run(testType+"__"+test.etlName, func(t *testing.T) {
 				msg := etl.InitCodeMsg{
 					InitMsgBase: etl.InitMsgBase{
-						EtlName:   test.etlName,
-						CommTypeX: test.commType,
-						Timeout:   etlBucketTimeout,
+						EtlName:     test.etlName,
+						CommTypeX:   test.commType,
+						InitTimeout: etlBucketTimeout,
 					},
 					Code:      []byte(test.code),
 					Deps:      []byte(test.deps),
@@ -1054,9 +1054,9 @@ func TestETLPodInitSpecFailure(t *testing.T) {
 
 			msg := &etl.InitSpecMsg{
 				InitMsgBase: etl.InitMsgBase{
-					EtlName:   test.etlName,
-					CommTypeX: test.commType,
-					Timeout:   failureTestTimeout,
+					EtlName:     test.etlName,
+					CommTypeX:   test.commType,
+					InitTimeout: failureTestTimeout,
 				},
 				Spec: spec,
 			}
@@ -1125,9 +1125,9 @@ def not_transform_func(input_bytes):
 			tools.CheckSkip(t, &tools.SkipTestArgs{Long: test.onlyLong})
 			msg := etl.InitCodeMsg{
 				InitMsgBase: etl.InitMsgBase{
-					EtlName:   test.etlName,
-					CommTypeX: test.commType,
-					Timeout:   failureTestTimeout,
+					EtlName:     test.etlName,
+					CommTypeX:   test.commType,
+					InitTimeout: failureTestTimeout,
 				},
 				Code:    []byte(test.code),
 				Deps:    []byte(test.deps),
@@ -1191,9 +1191,9 @@ def transform(input_bytes):
 		t.Run(test.etlName, func(t *testing.T) {
 			msg := etl.InitCodeMsg{
 				InitMsgBase: etl.InitMsgBase{
-					EtlName:   test.etlName,
-					CommTypeX: etl.Hpush, // TODO: enable runtime error retrieval for hpull in inline transform calls
-					Timeout:   etlBucketTimeout,
+					EtlName:     test.etlName,
+					CommTypeX:   etl.Hpush, // TODO: enable runtime error retrieval for hpull in inline transform calls
+					InitTimeout: etlBucketTimeout,
 				},
 				Code:    []byte(strings.Replace(failureTransformFunc, "<EXIT_CODE>", exitCode, 1)),
 				Runtime: test.runtime,
diff --git a/ais/tgtetl.go b/ais/tgtetl.go
@@ -188,7 +188,7 @@ func (t *target) inlineETL(w http.ResponseWriter, r *http.Request, dpq *dpq, lom
 	// NOTE:
 	// - poll for a while here for a possible abort error (e.g., pod runtime error)
 	// - and notice hardcoded timeout
-	if abortErr := xetl.AbortedAfter(etl.DefaultReqTimeout); abortErr != nil {
+	if abortErr := xetl.AbortedAfter(etl.DefaultObjTimeout); abortErr != nil {
 		t.writeErr(w, r, abortErr, ecode)
 		return
 	}
diff --git a/ext/etl/api.go b/ext/etl/api.go
@@ -44,8 +44,8 @@ const (
 )
 
 const (
-	DefaultTimeout    = 45 * time.Second
-	DefaultReqTimeout = 10 * time.Second
+	DefaultInitTimeout = 45 * time.Second
+	DefaultObjTimeout  = 10 * time.Second
 )
 
 // enum ETL lifecycle status (see docs/etl.md#etl-pod-lifecycle for details)
@@ -97,8 +97,9 @@ type (
 		EtlName          string       `json:"id"`
 		CommTypeX        string       `json:"communication"` // enum commTypes
 		ArgTypeX         string       `json:"argument"`      // enum argTypes
-		Timeout          cos.Duration `json:"timeout"`
-		SupportDirectPut bool         `json:"support_direct_put"`
+		InitTimeout      cos.Duration `json:"init_timeout,omitempty"`
+		ObjTimeout       cos.Duration `json:"obj_timeout,omitempty"`
+		SupportDirectPut bool         `json:"support_direct_put,omitempty"`
 	}
 	InitSpecMsg struct {
 		Spec []byte `json:"spec"`
@@ -190,11 +191,11 @@ func (*InitCodeMsg) MsgType() string    { return Code }
 func (*InitSpecMsg) MsgType() string    { return Spec }
 
 func (m *InitCodeMsg) String() string {
-	return fmt.Sprintf("init-%s[%s-%s-%s-%s]", Code, m.EtlName, m.CommTypeX, m.ArgTypeX, m.Runtime)
+	return fmt.Sprintf("init-%s[%s-%s-%s-%s], init-timeout=%s, object-timeout=%s", Code, m.EtlName, m.CommTypeX, m.ArgTypeX, m.Runtime, m.InitTimeout.D(), m.ObjTimeout.D())
 }
 
 func (m *InitSpecMsg) String() string {
-	return fmt.Sprintf("init-%s[%s-%s-%s]", Spec, m.EtlName, m.CommTypeX, m.ArgTypeX)
+	return fmt.Sprintf("init-%s[%s-%s-%s], init-timeout=%s, object-timeout=%s", Spec, m.EtlName, m.CommTypeX, m.ArgTypeX, m.InitTimeout.D(), m.ObjTimeout.D())
 }
 
 func (m *InitSpecMsg) errInvalidArg() error {
@@ -280,8 +281,11 @@ func (m *InitMsgBase) validate(detail string) error {
 		return cmn.NewErrUnsuppErr(err)
 	}
 	// NOTE: default timeout
-	if m.Timeout == 0 {
-		m.Timeout = cos.Duration(DefaultTimeout)
+	if m.InitTimeout == 0 {
+		m.InitTimeout = cos.Duration(DefaultInitTimeout)
+	}
+	if m.ObjTimeout == 0 {
+		m.ObjTimeout = cos.Duration(DefaultObjTimeout)
 	}
 	return nil
 }
diff --git a/ext/etl/boot.go b/ext/etl/boot.go
@@ -194,7 +194,7 @@ func (b *etlBootstrapper) createEntity(entity string) error {
 // NOTE: currently, we do require readinessProbe config in the ETL spec.
 func (b *etlBootstrapper) waitPodReady(podCtx context.Context) error {
 	var (
-		timeout     = b.msg.Timeout.D()
+		timeout     = b.msg.InitTimeout.D()
 		interval    = cos.ProbingFrequency(timeout)
 		client, err = k8s.GetClient()
 	)
diff --git a/ext/etl/comm_internal_test.go b/ext/etl/comm_internal_test.go
@@ -139,8 +139,8 @@ var _ = Describe("CommunicatorTest", func() {
 				boot := &etlBootstrapper{
 					msg: InitSpecMsg{
 						InitMsgBase: InitMsgBase{
-							CommTypeX: commType,
-							Timeout:   cos.Duration(DefaultTimeout),
+							CommTypeX:   commType,
+							InitTimeout: cos.Duration(DefaultInitTimeout),
 						},
 					},
 					pod:  pod,
diff --git a/ext/etl/communicator.go b/ext/etl/communicator.go
@@ -192,7 +192,7 @@ func handleRespEcode(ecode int, oah cos.OAH, r cos.ReadOpenCloser, err error) co
 
 func doWithTimeout(reqArgs *cmn.HreqArgs, getBody getBodyFunc, timeout time.Duration, started int64) (r cos.ReadCloseSizer, ecode int, err error) {
 	if timeout == 0 {
-		timeout = DefaultReqTimeout
+		timeout = DefaultObjTimeout
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 
@@ -225,7 +225,7 @@ func doWithTimeout(reqArgs *cmn.HreqArgs, getBody getBodyFunc, timeout time.Dura
 // pushComm: implements (Hpush | HpushStdin)
 //////////////
 
-func (pc *pushComm) doRequest(lom *core.LOM, timeout time.Duration, targs string, latestVer, sync bool, gargs *core.GetROCArgs) core.ReadResp {
+func (pc *pushComm) doRequest(lom *core.LOM, targs string, latestVer, sync bool, gargs *core.GetROCArgs) core.ReadResp {
 	if err := lom.InitBck(lom.Bucket()); err != nil {
 		return core.ReadResp{Err: err}
 	}
@@ -280,7 +280,7 @@ func (pc *pushComm) doRequest(lom *core.LOM, timeout time.Duration, targs string
 	}
 
 	// note: `Content-Length` header is set during `retryer.call()` below
-	r, ecode, err := doWithTimeout(reqArgs, getBody, timeout, started)
+	r, ecode, err := doWithTimeout(reqArgs, getBody, pc.boot.msg.ObjTimeout.D(), started)
 	if err != nil {
 		return core.ReadResp{Err: err, Ecode: ecode}
 	}
@@ -289,7 +289,7 @@ func (pc *pushComm) doRequest(lom *core.LOM, timeout time.Duration, targs string
 }
 
 func (pc *pushComm) InlineTransform(w http.ResponseWriter, _ *http.Request, lom *core.LOM, latestVer bool, targs string) (int, error) {
-	resp := pc.doRequest(lom, pc.boot.msg.Timeout.D(), targs, latestVer, false, nil)
+	resp := pc.doRequest(lom, targs, latestVer, false, nil)
 	if resp.Err != nil {
 		return resp.Ecode, resp.Err
 	}
@@ -310,7 +310,7 @@ func (pc *pushComm) InlineTransform(w http.ResponseWriter, _ *http.Request, lom
 }
 
 func (pc *pushComm) OfflineTransform(lom *core.LOM, latestVer, sync bool, gargs *core.GetROCArgs) core.ReadResp {
-	resp := pc.doRequest(lom, pc.boot.msg.Timeout.D(), "", latestVer, sync, gargs)
+	resp := pc.doRequest(lom, "", latestVer, sync, gargs)
 	if cmn.Rom.FastV(5, cos.SmoduleETL) {
 		nlog.Infoln(Hpush, lom.Cname(), resp.Err, resp.Ecode)
 	}
@@ -387,7 +387,7 @@ func (rc *redirectComm) OfflineTransform(lom *core.LOM, latestVer, _ bool, gargs
 		reqArgs.Header.Add(apc.HdrNodeURL, gargs.Daddr)
 	}
 
-	r, ecode, err := doWithTimeout(reqArgs, nil, rc.boot.msg.Timeout.D(), started)
+	r, ecode, err := doWithTimeout(reqArgs, nil, rc.boot.msg.ObjTimeout.D(), started)
 	if err != nil {
 		return core.ReadResp{Err: err, Ecode: ecode}
 	}
diff --git a/ext/etl/transform.go b/ext/etl/transform.go
@@ -317,9 +317,7 @@ func start(msg *InitSpecMsg, xid, secret string, opts StartOpts, config *cmn.Con
 		goto cleanup
 	}
 
-	if cmn.Rom.FastV(4, cos.SmoduleETL) {
-		nlog.Infof("pod %q is running, %+v, %s", podName, msg, boot.errCtx)
-	}
+	nlog.Infof("pod %q is running, %+v, %s", podName, msg, boot.errCtx)
 
 	return podName, svcName, xctn, nil
 
@@ -547,8 +545,8 @@ func deleteEntity(errCtx *cmn.ETLErrCtx, entityType, entityName string) error {
 	}
 
 	// wait
-	interval := cos.ProbingFrequency(DefaultTimeout)
-	err = wait.PollUntilContextTimeout(context.Background(), interval, DefaultTimeout, false, /*immediate*/
+	interval := cos.ProbingFrequency(DefaultInitTimeout)
+	err = wait.PollUntilContextTimeout(context.Background(), interval, DefaultInitTimeout, false, /*immediate*/
 		func(context.Context) (done bool, err error) {
 			var exists bool
 			exists, err = client.CheckExists(entityType, entityName)
diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md
@@ -8,6 +8,10 @@ We structure this changelog in accordance with [Keep a Changelog](https://keepac
 
 ## Unreleased
 
+### Added
+
+- Add `obj_timeout` and `init_timeout` fields in `InitETLArgs` type to align with updated API fields.
+
 ## [1.13.7] - 2025-05-06
 
 ### Added
diff --git a/python/aistore/sdk/etl/etl.py b/python/aistore/sdk/etl/etl.py
@@ -22,6 +22,7 @@
     DEFAULT_ETL_RUNTIME,
     DEFAULT_ETL_COMM,
     DEFAULT_ETL_TIMEOUT,
+    DEFAULT_ETL_OBJ_TIMEOUT,
     ETL_COMM_SPEC,
     ETL_COMM_CODE,
     CODE_TEMPLATE,
@@ -65,11 +66,13 @@ def name(self) -> str:
         """Name of the ETL"""
         return self._name
 
+    # pylint: disable=too-many-arguments,too-many-positional-arguments
     def init_spec(
         self,
         template: str,
         communication_type: str = DEFAULT_ETL_COMM,
-        timeout: str = DEFAULT_ETL_TIMEOUT,
+        init_timeout: str = DEFAULT_ETL_TIMEOUT,
+        obj_timeout: str = DEFAULT_ETL_OBJ_TIMEOUT,
         arg_type: str = "",
     ) -> str:
         """
@@ -80,7 +83,8 @@ def init_spec(
                 Existing templates can be found at `sdk.etl_templates`
                 For more information visit: https://github.com/NVIDIA/ais-etl/tree/main/transformers
             communication_type (str): Communication type of the ETL (options: hpull, hpush)
-            timeout (str): Timeout of the ETL job (e.g. 5m for 5 minutes). Defaults to 5m.
+            init_timeout (str): [optional, default="5m"] Timeout of the ETL job (e.g. 5m for 5 minutes)
+            obj_timeout (str): [optional, default="45s"] Timeout of transforming a single object
         Returns:
             Job ID string associated with this ETL
         """
@@ -95,14 +99,15 @@ def init_spec(
             spec=spec_encoded,
             etl_name=self._name,
             communication_type=communication_type,
-            timeout=timeout,
+            init_timeout=init_timeout,
+            obj_timeout=obj_timeout,
             arg_type=arg_type,
         ).as_dict()
 
         return self._client.request(
             HTTP_METHOD_PUT,
             path=URL_PATH_ETL,
-            timeout=convert_to_seconds(timeout),
+            timeout=convert_to_seconds(init_timeout),
             json=value,
         ).text
 
@@ -114,7 +119,8 @@ def init_code(
         preimported_modules: List[str] = None,
         runtime: str = _get_default_runtime(),
         communication_type: str = DEFAULT_ETL_COMM,
-        timeout: str = DEFAULT_ETL_TIMEOUT,
+        init_timeout: str = DEFAULT_ETL_TIMEOUT,
+        obj_timeout: str = DEFAULT_ETL_OBJ_TIMEOUT,
         chunk_size: int = None,
         arg_type: str = "",
     ) -> str:
@@ -131,7 +137,8 @@ def init_code(
                 python3.12v2, python3.13v2] (see ext/etl/runtime/all.go)
             communication_type (str): [optional, default="hpush"] Communication type of the ETL (options: hpull,
                 hpush, io)
-            timeout (str): [optional, default="5m"] Timeout of the ETL job (e.g. 5m for 5 minutes)
+            init_timeout (str): [optional, default="5m"] Timeout of the ETL job (e.g. 5m for 5 minutes)
+            obj_timeout (str): [optional, default="45s"] Timeout of transforming a single object
             chunk_size (int): Chunk size in bytes if transform function in streaming data.
                 (whole object is read by default)
             arg_type (optional, str): The type of argument the runtime will provide the transform function.
@@ -151,7 +158,8 @@ def init_code(
             etl_name=self._name,
             runtime=runtime,
             communication_type=communication_type,
-            timeout=timeout,
+            init_timeout=init_timeout,
+            obj_timeout=obj_timeout,
             dependencies=self._encode_dependencies(dependencies),
             functions=functions,
             code=self._encode_transform(
@@ -164,7 +172,7 @@ def init_code(
         return self._client.request(
             HTTP_METHOD_PUT,
             path=URL_PATH_ETL,
-            timeout=convert_to_seconds(timeout),
+            timeout=convert_to_seconds(init_timeout),
             json=value,
         ).text
 
diff --git a/python/aistore/sdk/etl/etl_const.py b/python/aistore/sdk/etl/etl_const.py
@@ -1,6 +1,7 @@
 # Defaults
 DEFAULT_ETL_COMM = "hpush"
 DEFAULT_ETL_TIMEOUT = "5m"
+DEFAULT_ETL_OBJ_TIMEOUT = "45s"
 DEFAULT_ETL_RUNTIME = "python3.13v2"
 
 # ETL comm types
diff --git a/python/aistore/sdk/types.py b/python/aistore/sdk/types.py
@@ -318,7 +318,8 @@ class ETLDetails(BaseModel):
 
     id: str
     communication: str
-    timeout: str
+    init_timeout: Optional[str]
+    obj_timeout: Optional[str]
     code: Optional[bytes]
     spec: Optional[str]
     dependencies: Optional[str]
@@ -346,7 +347,8 @@ class InitETLArgs(BaseModel):
 
     etl_name: str
     communication_type: str
-    timeout: str
+    init_timeout: Optional[str]
+    obj_timeout: Optional[str]
     arg_type: str = ""
 
 
@@ -360,7 +362,8 @@ class InitSpecETLArgs(InitETLArgs):
     def as_dict(self):
         return {
             "id": self.etl_name,
-            "timeout": self.timeout,
+            "init_timeout": self.init_timeout,
+            "obj_timeout": self.obj_timeout,
             "communication": f"{self.communication_type}://",
             "spec": self.spec,
             "argument": self.arg_type,
@@ -383,7 +386,8 @@ def as_dict(self):
             "id": self.etl_name,
             "runtime": self.runtime,
             "communication": f"{self.communication_type}://",
-            "timeout": self.timeout,
+            "init_timeout": self.init_timeout,
+            "obj_timeout": self.obj_timeout,
             "funcs": self.functions,
             "code": self.code,
             "dependencies": self.dependencies,
diff --git a/python/tests/unit/sdk/test_etl.py b/python/tests/unit/sdk/test_etl.py
diff --git a/tools/tetl/etl.go b/tools/tetl/etl.go

Original file line number	Diff line number	Diff line change
`@@ -188,7 +188,7 @@ func (t target) inlineETL(w http.ResponseWriter, r http.Request, dpq *dpq, lom`
`188`	`188`	`// NOTE:`
`189`	`189`	`// - poll for a while here for a possible abort error (e.g., pod runtime error)`
`190`	`190`	`// - and notice hardcoded timeout`
`191`		`- if abortErr := xetl.AbortedAfter(etl.DefaultReqTimeout); abortErr != nil {`
	`191`	`+ if abortErr := xetl.AbortedAfter(etl.DefaultObjTimeout); abortErr != nil {`
`192`	`192`	`t.writeErr(w, r, abortErr, ecode)`
`193`	`193`	`return`
`194`	`194`	`}`
Original file line number	Diff line number	Diff line change
`@@ -44,8 +44,8 @@ const (`
`44`	`44`	`)`
`45`	`45`
`46`	`46`	`const (`
`47`		`- DefaultTimeout = 45 * time.Second`
`48`		`- DefaultReqTimeout = 10 * time.Second`
	`47`	`+ DefaultInitTimeout = 45 * time.Second`
	`48`	`+ DefaultObjTimeout = 10 * time.Second`
`49`	`49`	`)`
`50`	`50`
`51`	`51`	`// enum ETL lifecycle status (see docs/etl.md#etl-pod-lifecycle for details)`
`@@ -97,8 +97,9 @@ type (`
`97`	`97`	EtlName string `json:"id"`
`98`	`98`	CommTypeX string `json:"communication"` // enum commTypes
`99`	`99`	ArgTypeX string `json:"argument"` // enum argTypes
`100`		- Timeout cos.Duration `json:"timeout"`
`101`		- SupportDirectPut bool `json:"support_direct_put"`
	`100`	+ InitTimeout cos.Duration `json:"init_timeout,omitempty"`
	`101`	+ ObjTimeout cos.Duration `json:"obj_timeout,omitempty"`
	`102`	+ SupportDirectPut bool `json:"support_direct_put,omitempty"`
`102`	`103`	`}`
`103`	`104`	`InitSpecMsg struct {`
`104`	`105`	Spec []byte `json:"spec"`
`@@ -190,11 +191,11 @@ func (*InitCodeMsg) MsgType() string { return Code }`
`190`	`191`	`func (*InitSpecMsg) MsgType() string { return Spec }`
`191`	`192`
`192`	`193`	`func (m *InitCodeMsg) String() string {`
`193`		`- return fmt.Sprintf("init-%s[%s-%s-%s-%s]", Code, m.EtlName, m.CommTypeX, m.ArgTypeX, m.Runtime)`
	`194`	`+ return fmt.Sprintf("init-%s[%s-%s-%s-%s], init-timeout=%s, object-timeout=%s", Code, m.EtlName, m.CommTypeX, m.ArgTypeX, m.Runtime, m.InitTimeout.D(), m.ObjTimeout.D())`
`194`	`195`	`}`
`195`	`196`
`196`	`197`	`func (m *InitSpecMsg) String() string {`
`197`		`- return fmt.Sprintf("init-%s[%s-%s-%s]", Spec, m.EtlName, m.CommTypeX, m.ArgTypeX)`
	`198`	`+ return fmt.Sprintf("init-%s[%s-%s-%s], init-timeout=%s, object-timeout=%s", Spec, m.EtlName, m.CommTypeX, m.ArgTypeX, m.InitTimeout.D(), m.ObjTimeout.D())`
`198`	`199`	`}`
`199`	`200`
`200`	`201`	`func (m *InitSpecMsg) errInvalidArg() error {`
`@@ -280,8 +281,11 @@ func (m *InitMsgBase) validate(detail string) error {`
`280`	`281`	`return cmn.NewErrUnsuppErr(err)`
`281`	`282`	`}`
`282`	`283`	`// NOTE: default timeout`
`283`		`- if m.Timeout == 0 {`
`284`		`- m.Timeout = cos.Duration(DefaultTimeout)`
	`284`	`+ if m.InitTimeout == 0 {`
	`285`	`+ m.InitTimeout = cos.Duration(DefaultInitTimeout)`
	`286`	`+ }`
	`287`	`+ if m.ObjTimeout == 0 {`
	`288`	`+ m.ObjTimeout = cos.Duration(DefaultObjTimeout)`
`285`	`289`	`}`
`286`	`290`	`return nil`
`287`	`291`	`}`
Original file line number	Diff line number	Diff line change
`@@ -194,7 +194,7 @@ func (b *etlBootstrapper) createEntity(entity string) error {`
`194`	`194`	`// NOTE: currently, we do require readinessProbe config in the ETL spec.`
`195`	`195`	`func (b *etlBootstrapper) waitPodReady(podCtx context.Context) error {`
`196`	`196`	`var (`
`197`		`- timeout = b.msg.Timeout.D()`
	`197`	`+ timeout = b.msg.InitTimeout.D()`
`198`	`198`	`interval = cos.ProbingFrequency(timeout)`
`199`	`199`	`client, err = k8s.GetClient()`
`200`	`200`	`)`
Original file line number	Diff line number	Diff line change
`@@ -192,7 +192,7 @@ func handleRespEcode(ecode int, oah cos.OAH, r cos.ReadOpenCloser, err error) co`
`192`	`192`
`193`	`193`	`func doWithTimeout(reqArgs *cmn.HreqArgs, getBody getBodyFunc, timeout time.Duration, started int64) (r cos.ReadCloseSizer, ecode int, err error) {`
`194`	`194`	`if timeout == 0 {`
`195`		`- timeout = DefaultReqTimeout`
	`195`	`+ timeout = DefaultObjTimeout`
`196`	`196`	`}`
`197`	`197`	`ctx, cancel := context.WithTimeout(context.Background(), timeout)`
`198`	`198`
`@@ -225,7 +225,7 @@ func doWithTimeout(reqArgs *cmn.HreqArgs, getBody getBodyFunc, timeout time.Dura`
`225`	`225`	`// pushComm: implements (Hpush \| HpushStdin)`
`226`	`226`	`//////////////`
`227`	`227`
`228`		`-func (pc pushComm) doRequest(lom core.LOM, timeout time.Duration, targs string, latestVer, sync bool, gargs *core.GetROCArgs) core.ReadResp {`
	`228`	`+func (pc pushComm) doRequest(lom core.LOM, targs string, latestVer, sync bool, gargs *core.GetROCArgs) core.ReadResp {`
`229`	`229`	`if err := lom.InitBck(lom.Bucket()); err != nil {`
`230`	`230`	`return core.ReadResp{Err: err}`
`231`	`231`	`}`
`@@ -280,7 +280,7 @@ func (pc pushComm) doRequest(lom core.LOM, timeout time.Duration, targs string`
`280`	`280`	`}`
`281`	`281`
`282`	`282`	// note: `Content-Length` header is set during `retryer.call()` below
`283`		`- r, ecode, err := doWithTimeout(reqArgs, getBody, timeout, started)`
	`283`	`+ r, ecode, err := doWithTimeout(reqArgs, getBody, pc.boot.msg.ObjTimeout.D(), started)`
`284`	`284`	`if err != nil {`
`285`	`285`	`return core.ReadResp{Err: err, Ecode: ecode}`
`286`	`286`	`}`
`@@ -289,7 +289,7 @@ func (pc pushComm) doRequest(lom core.LOM, timeout time.Duration, targs string`
`289`	`289`	`}`
`290`	`290`
`291`	`291`	`func (pc pushComm) InlineTransform(w http.ResponseWriter, _ http.Request, lom *core.LOM, latestVer bool, targs string) (int, error) {`
`292`		`- resp := pc.doRequest(lom, pc.boot.msg.Timeout.D(), targs, latestVer, false, nil)`
	`292`	`+ resp := pc.doRequest(lom, targs, latestVer, false, nil)`
`293`	`293`	`if resp.Err != nil {`
`294`	`294`	`return resp.Ecode, resp.Err`
`295`	`295`	`}`
`@@ -310,7 +310,7 @@ func (pc pushComm) InlineTransform(w http.ResponseWriter, _ http.Request, lom`
`310`	`310`	`}`
`311`	`311`
`312`	`312`	`func (pc pushComm) OfflineTransform(lom core.LOM, latestVer, sync bool, gargs *core.GetROCArgs) core.ReadResp {`
`313`		`- resp := pc.doRequest(lom, pc.boot.msg.Timeout.D(), "", latestVer, sync, gargs)`
	`313`	`+ resp := pc.doRequest(lom, "", latestVer, sync, gargs)`
`314`	`314`	`if cmn.Rom.FastV(5, cos.SmoduleETL) {`
`315`	`315`	`nlog.Infoln(Hpush, lom.Cname(), resp.Err, resp.Ecode)`
`316`	`316`	`}`
`@@ -387,7 +387,7 @@ func (rc redirectComm) OfflineTransform(lom core.LOM, latestVer, _ bool, gargs`
`387`	`387`	`reqArgs.Header.Add(apc.HdrNodeURL, gargs.Daddr)`
`388`	`388`	`}`
`389`	`389`
`390`		`- r, ecode, err := doWithTimeout(reqArgs, nil, rc.boot.msg.Timeout.D(), started)`
	`390`	`+ r, ecode, err := doWithTimeout(reqArgs, nil, rc.boot.msg.ObjTimeout.D(), started)`
`391`	`391`	`if err != nil {`
`392`	`392`	`return core.ReadResp{Err: err, Ecode: ecode}`
`393`	`393`	`}`