[debezium] Check for negative overflow when coercing ints (#273)

artie-labs · Mar 11, 2024 · 0c519fa · 0c519fa
1 parent ded87b8
commit 0c519fa
Show file tree

Hide file tree

Showing 4 changed files with 84 additions and 29 deletions.
diff --git a/lib/debezium/converters/int.go b/lib/debezium/converters/int.go
@@ -10,18 +10,18 @@ func asInt16(value any) (int16, error) {
 	case int16:
 		return castValue, nil
 	case int32:
-		if castValue > math.MaxInt16 {
-			return 0, fmt.Errorf("value is too large for int16")
+		if castValue > math.MaxInt16 || castValue < math.MinInt16 {
+			return 0, fmt.Errorf("value overflows int16")
 		}
 		return int16(castValue), nil
 	case int:
-		if castValue > math.MaxInt16 {
-			return 0, fmt.Errorf("value is too large for int16")
+		if castValue > math.MaxInt16 || castValue < math.MinInt16 {
+			return 0, fmt.Errorf("value overflows int16")
 		}
 		return int16(castValue), nil
 	case int64:
-		if castValue > math.MaxInt16 {
-			return 0, fmt.Errorf("value is too large for int16")
+		if castValue > math.MaxInt16 || castValue < math.MinInt16 {
+			return 0, fmt.Errorf("value overflows int16")
 		}
 		return int16(castValue), nil
 	}
@@ -35,13 +35,13 @@ func asInt32(value any) (int32, error) {
 	case int32:
 		return castValue, nil
 	case int:
-		if castValue > math.MaxInt32 {
-			return 0, fmt.Errorf("value is too large for int32")
+		if castValue > math.MaxInt32 || castValue < math.MinInt32 {
+			return 0, fmt.Errorf("value overflows int32")
 		}
 		return int32(castValue), nil
 	case int64:
-		if castValue > math.MaxInt32 {
-			return 0, fmt.Errorf("value is too large for int32")
+		if castValue > math.MaxInt32 || castValue < math.MinInt32 {
+			return 0, fmt.Errorf("value overflows int32")
 		}
 		return int32(castValue), nil
 	}

diff --git a/lib/debezium/converters/int_test.go b/lib/debezium/converters/int_test.go
@@ -24,15 +24,26 @@ func TestAsInt16(t *testing.T) {
 		assert.Equal(t, int16(1234), value)
 	}
 	{
-		// int32 - just large enough
+		// int32 - just negative enough
+		value, err := asInt16(int32(math.MinInt16))
+		assert.NoError(t, err)
+		assert.Equal(t, int16(math.MinInt16), value)
+	}
+	{
+		// int32 - negative overflow
+		_, err := asInt16(int32(math.MinInt16 - 1))
+		assert.ErrorContains(t, err, "value overflows int16")
+	}
+	{
+		// int32 - just positive enough
 		value, err := asInt16(int32(math.MaxInt16))
 		assert.NoError(t, err)
 		assert.Equal(t, int16(math.MaxInt16), value)
 	}
 	{
-		// int32 - too large
+		// int32 - positive overflow
 		_, err := asInt16(int32(math.MaxInt16 + 1))
-		assert.ErrorContains(t, err, "value is too large for int16")
+		assert.ErrorContains(t, err, "value overflows int16")
 	}
 	{
 		// int64
@@ -41,15 +52,26 @@ func TestAsInt16(t *testing.T) {
 		assert.Equal(t, int16(1234), value)
 	}
 	{
-		// int64 - just large enough
+		// int64 - just negative enough
+		value, err := asInt16(int64(math.MinInt16))
+		assert.NoError(t, err)
+		assert.Equal(t, int16(math.MinInt16), value)
+	}
+	{
+		// int64 - negative overflow
+		_, err := asInt16(int64(math.MinInt16 - 1))
+		assert.ErrorContains(t, err, "value overflows int16")
+	}
+	{
+		// int64 - just positive enough
 		value, err := asInt16(int64(math.MaxInt16))
 		assert.NoError(t, err)
 		assert.Equal(t, int16(math.MaxInt16), value)
 	}
 	{
-		// int64 - too large
+		// int64 - positive overflow
 		_, err := asInt16(int64(math.MaxInt16 + 1))
-		assert.ErrorContains(t, err, "value is too large for int16")
+		assert.ErrorContains(t, err, "value overflows int16")
 	}
 	{
 		// int
@@ -58,15 +80,26 @@ func TestAsInt16(t *testing.T) {
 		assert.Equal(t, int16(1234), value)
 	}
 	{
-		// int - just large enough
+		// int - just negative enough
+		value, err := asInt16(int(math.MinInt16))
+		assert.NoError(t, err)
+		assert.Equal(t, int16(math.MinInt16), value)
+	}
+	{
+		// int - negative overflow
+		_, err := asInt16(int(math.MinInt16 - 1))
+		assert.ErrorContains(t, err, "value overflows int16")
+	}
+	{
+		// int - just positive enough
 		value, err := asInt16(int(math.MaxInt16))
 		assert.NoError(t, err)
 		assert.Equal(t, int16(math.MaxInt16), value)
 	}
 	{
-		// int - too large
+		// int - positive overflow
 		_, err := asInt16(int(math.MaxInt16 + 1))
-		assert.ErrorContains(t, err, "value is too large for int16")
+		assert.ErrorContains(t, err, "value overflows int16")
 	}
 }
 
@@ -94,15 +127,26 @@ func TestAsInt32(t *testing.T) {
 		assert.Equal(t, int32(1234), value)
 	}
 	{
-		// int64 - just large enough
+		// int64 - just negative enough
+		value, err := asInt32(int64(math.MinInt32))
+		assert.NoError(t, err)
+		assert.Equal(t, int32(math.MinInt32), value)
+	}
+	{
+		// int64 - negative overflow
+		_, err := asInt32(int64(math.MinInt32 - 1))
+		assert.ErrorContains(t, err, "value overflows int32")
+	}
+	{
+		// int64 - just positive enough
 		value, err := asInt32(int64(math.MaxInt32))
 		assert.NoError(t, err)
 		assert.Equal(t, int32(math.MaxInt32), value)
 	}
 	{
-		// int64 - too large
+		// int64 - positive overflow
 		_, err := asInt32(int64(math.MaxInt32 + 1))
-		assert.ErrorContains(t, err, "value is too large for int32")
+		assert.ErrorContains(t, err, "value overflows int32")
 	}
 	{
 		// int
@@ -111,15 +155,26 @@ func TestAsInt32(t *testing.T) {
 		assert.Equal(t, int32(1234), value)
 	}
 	{
-		// int - just large enough
+		// int - just negative enough
+		value, err := asInt32(int(math.MinInt32))
+		assert.NoError(t, err)
+		assert.Equal(t, int32(math.MinInt32), value)
+	}
+	{
+		// int - negative overflow
+		_, err := asInt32(int(math.MinInt32 - 1))
+		assert.ErrorContains(t, err, "value overflows int32")
+	}
+	{
+		// int - just postive enough
 		value, err := asInt32(int(math.MaxInt32))
 		assert.NoError(t, err)
 		assert.Equal(t, int32(math.MaxInt32), value)
 	}
 	{
-		// int - too large
+		// int - positive overflow
 		_, err := asInt32(int(math.MaxInt32 + 1))
-		assert.ErrorContains(t, err, "value is too large for int32")
+		assert.ErrorContains(t, err, "value overflows int32")
 	}
 }
 

diff --git a/lib/debezium/converters/time_test.go b/lib/debezium/converters/time_test.go
@@ -176,6 +176,6 @@ func TestYearConverter_Convert(t *testing.T) {
 	{
 		// int64 - too big
 		_, err := converter.Convert(int64(math.MaxInt32 + 1))
-		assert.ErrorContains(t, err, "value is too large for int32")
+		assert.ErrorContains(t, err, "value overflows int32")
 	}
 }
diff --git a/sources/postgres/adapter/converters_test.go b/sources/postgres/adapter/converters_test.go
@@ -128,18 +128,18 @@ func TestPgIntervalConverter_Convert(t *testing.T) {
 		assert.ErrorContains(t, err, "positive microseconds are too large for an int64")
 	}
 	{
-		// Valid pgtype.Interval - very large but no underflow
+		// Valid pgtype.Interval - very large but no overflow
 		value, err := converter.Convert(pgtype.Interval{Valid: true, Months: -292_000 * 12})
 		assert.NoError(t, err)
 		assert.Equal(t, int64(-9_214_819_200_000_000_000), value)
 	}
 	{
-		// Valid pgtype.Interval - underflow
+		// Valid pgtype.Interval - negative overflow
 		_, err := converter.Convert(pgtype.Interval{Valid: true, Months: -293_000 * 12})
 		assert.ErrorContains(t, err, "negative microseconds are too large for an int64")
 	}
 	{
-		// Valid pgtype.Interval - underflow
+		// Valid pgtype.Interval - positive overflow
 		_, err := converter.Convert(pgtype.Interval{Valid: true, Microseconds: math.MinInt64, Days: -1})
 		assert.ErrorContains(t, err, "negative microseconds are too large for an int64")
 	}